Agents

The CoCo agent runs cortex exec inside a Docker sandbox.

inspect_coco.agents.coco

Cortex Code agent for Inspect AI — runs cortex CLI in Docker sandbox.

coco(timeout_sec=900, max_turns=None, remove_skills=None, model_name=None, connection_name=None, workdir=None)

CoCo agent — runs cortex exec in a Docker sandbox.

Uses the CI/CD-optimized cortex exec command (beta) which runs non-interactively with plan mode disabled and interactive prompts auto-rejected. Model defaults to CoCo auto mode unless overridden.

Parameters:

Name	Type	Description	Default
timeout_sec	int	Maximum execution time for the cortex CLI.	900
max_turns	int | None	Maximum agentic turns (safety ceiling).	None
remove_skills	list[str] | None	Bundled skills to disable.	None
model_name	str | None	Model override (e.g. "claude-sonnet-4-5"). Default: auto mode.	None
connection_name	str | None	Named Snowflake connection to use.	None
workdir	str | None	Working directory inside the sandbox.	None

Source code in src/inspect_coco/agents/coco.py

@agent
def coco(
    timeout_sec: int = 900,
    max_turns: int | None = None,
    remove_skills: list[str] | None = None,
    model_name: str | None = None,
    connection_name: str | None = None,
    workdir: str | None = None,
) -> Agent:
    """CoCo agent — runs `cortex exec` in a Docker sandbox.

    Uses the CI/CD-optimized `cortex exec` command (beta) which runs
    non-interactively with plan mode disabled and interactive prompts
    auto-rejected. Model defaults to CoCo auto mode unless overridden.

    Args:
        timeout_sec: Maximum execution time for the cortex CLI.
        max_turns: Maximum agentic turns (safety ceiling).
        remove_skills: Bundled skills to disable.
        model_name: Model override (e.g. "claude-sonnet-4-5"). Default: auto mode.
        connection_name: Named Snowflake connection to use.
        workdir: Working directory inside the sandbox.
    """
    _credentials_deployed = False
    _token_proxy = None

    async def execute(state: AgentState) -> AgentState:
        nonlocal _credentials_deployed, _token_proxy

        # Always resolve fresh credentials and deploy into the sandbox.
        # Each epoch gets a fresh container, so we must re-deploy every time.
        config = resolve_connection(connection_name)

        # Start token proxy for OAuth connections (once per agent lifecycle)
        if config.oauth_access_token and _token_proxy is None:
            from inspect_coco.proxy.server import TokenProxy

            _token_proxy = TokenProxy(account=config.account, role=config.role)
            _token_proxy.start()

        env_vars = await _deploy_to_sandbox(config, token_proxy=_token_proxy)

        # Resolve model: explicit param > env var > None (CLI default)
        resolved_model = model_name or os.environ.get("INSPECT_COCO_MODEL")

        # Extract prompt from the last user message
        prompt = _extract_prompt(state)

        # Write prompt to file in sandbox (avoids shell escaping issues with large markdown)
        prompt_path = "/tmp/inspect-coco-prompt.md"
        await sandbox().write_file(prompt_path, prompt)

        # Build cortex exec command (uses --file to read prompt)
        cmd = _build_command(
            prompt_file=prompt_path,
            model_name=resolved_model,
            max_turns=max_turns,
            connection_name=connection_name,
            workdir=workdir,
        )

        # Build environment
        cortex_channel = os.environ.get("INSPECT_COCO_CHANNEL", "stable")
        env = {
            **env_vars,
            "COCO_TELEMETRY_DISABLED": "true",
            "CORTEX_CHANNEL": cortex_channel,
        }
        if remove_skills:
            env["CORTEX_DISABLE_BUNDLED_SKILLS"] = ",".join(remove_skills)

        # Execute in sandbox
        logger.info("Running cortex CLI (timeout=%ds)", timeout_sec)
        result = await sandbox().exec(cmd=cmd, timeout=timeout_sec, env=env)

        # Parse trajectory from stream-json output
        trajectory = parse_stream_json(result.stdout, exit_code=result.returncode)

        if result.returncode != 0 and not trajectory.final_response:
            trajectory.final_response = (
                f"Agent exited with code {result.returncode}.\n"
                f"stderr: {result.stderr[:500] if result.stderr else 'none'}"
            )

        # Update agent state
        state.messages.append(ChatMessageAssistant(content=trajectory.final_response))
        state.output = ModelOutput.from_content(
            model="cortex-code",
            content=trajectory.final_response,
            stop_reason="stop" if result.returncode == 0 else "error",
        )

        return state

    return execute