Coding harnesses

Delegate a task from a traversal to a coding-agent CLI with app.harness. Choose a provider and working directory, set deadlines, validate the answer and inspect recorded progress. Local mode needs a supported CLI and its credentials on the worker host. Remote mode, described later, delegates to an ATF Harness service. Repository tests use fake executables or a protocol double rather than proving access to live providers.

For local mode, install a supported executable such as atfHar, claude, codex, opencode or gemini on the worker's PATH, and configure that CLI's own credentials.

When to reach for a harness

Use a harness when the task needs a coding agent's iterative tool loop: investigating a repository, fixing tests, editing files or generating a project. A harness can run commands and modify its working files when its tools permit that. The calling traversal receives the result and metrics, while internal actions remain within the harness.

For a model request or bounded tool loop, use app.ai. Its graph-backed tools produce separate child steps. This gives more detailed graph visibility without launching a separate coding-agent process.

Install a provider

Install the provider CLI on the worker host and configure its credentials there. The graph server does not need that executable.

# atf-harness: the family's own runtime, a Bun application
git clone <the atf-harness repository> && cd atf-harness && bun install && bun link
atfHar --version

# claude-code: Anthropic's CLI; signs in with `claude` login or ANTHROPIC_API_KEY
npm install -g @anthropic-ai/claude-code

# codex: OpenAI's CLI; signs in with `codex` login or OPENAI_API_KEY
npm install -g @openai/codex

# opencode: its installer script from opencode.ai; brings its own provider configuration
curl -fsSL https://opencode.ai/install | bash

Check executable availability before starting a paid call:

python -m examples.harness.python.call --doctor

harness_doctor() searches PATH without running the binaries. available: true establishes that an executable exists, not that it can authenticate. Missing or expired credentials fail later during invocation, with the CLI's own error, which may include a provider 401.

Not verified: these vendor installation commands against current Claude Code, Codex and OpenCode releases. The repository does not execute them in its tests.

What the bridge does

app.harness calls run_harness in harness_bridge.py. The bridge starts and controls the process without a shell, reads JSON-lines output, enforces time and output limits, and terminates the process group on cancellation or deadline. Providers in harness_providers.py build command arguments and parse events; they do not start processes themselves.

Setting cwd selects a working directory; it is not a security sandbox. The underlying CLI determines its filesystem and tool permissions. A nonzero exit, an error event or a missing assistant response raises HarnessFailure rather than producing an invented result.

A local invocation follows these steps:

1. Select the provider from the call's provider, the agent's harness_provider, ATFGRAPH_HARNESS_PROVIDER, or the default atf-harness, in that order.
2. Launch the CLI as a child process group in cwd, without a shell or inbound port. Apply the provider's tool settings and parse streamed events. Tool restrictions differ by provider, as shown below.
3. Enforce a default 300-second timeout and 4 MiB combined-output limit. An optional max_cost_usd uses reported costs; streamed overruns terminate the process with non-retryable harness_cost_exceeded.
4. If schema is supplied, add it to the prompt and validate the returned JSON, tolerating a Markdown code fence. Store the parsed object in result.parsed.
5. Return HarnessResult with text, provider, model, session, tokens, turns, duration and any provider-reported price.

The graph step is the traversal that called the harness. CLI events add progress summaries to that step, rather than creating child graph steps. Local harness usage is returned for your code to record; see What comes back. Use max_attempts=1 for write operations so an automatic retry does not repeat edits.

Native tool routing for local ATF Harness

NativeHarnessTools opts a local ATF Harness invocation into native graph tool steps. Register it before serving the agent; each invocation still uses the configured model and the harness's original tool implementations and approval checks:

from atfgraph import Agent
from atfgraph.harness import NativeHarnessTools

app = Agent(node_id="coder", version="1.0.0")
routed = NativeHarnessTools(app, state_dir="/private/agent-state/harness-tools")

@app.traversal(max_attempts=1, timeout_seconds=900)
async def code(prompt: str, work_dir: str, model: str | None = None) -> dict:
    result = await routed.harness(
        prompt, operation_key="coding", cwd=work_dir, model=model,
        tools=["read", "glob", "grep", "write", "edit", "bash"], timeout=600,
    )
    return {"text": result.text}

The path now contains coder.harness_read, coder.harness_write, and the other selected tool capabilities as children of code. Each operation has its own input, output, attempts and status. Arguments over 512 KiB stay in the shared private journal; the graph child records a digest reference and validates it before loading the original arguments. Results over 512 KiB become artifacts; the harness receives the original result. Protocol frames are capped at 4 MiB; oversized payloads fail rather than being truncated. Tool payloads are recorded, so keep credentials in the harness environment rather than tool arguments. ATFGRAPH_ environment variables are excluded from the harness subprocess.

The same invocation key and harness tool-call ID reuse the same graph child. Changing the input under an existing ID is refused. A private local journal saves intent before execution and results before graph completion. After a tool worker interruption, saved results are reused; a live executor can reattach to the already-running native operation. If a mutation has an intent but no recoverable result, it fails with harness_tool_outcome_unknown, without issuing it again. Interrupted read, glob, and grep may retry. max_operations defaults to 100 per invocation; each native tool has a 300-second execution timeout by default. Queueing and recovery waits use the enclosing harness deadline rather than that tool limit. Known native tool errors remain failed graph children and are returned to the model, which can correct the request and continue. Routing failures and uncertain outcomes stop the invocation. As with app.harness, omitting cwd or passing None uses the agent's harness_cwd.

This is operation-level recovery. A completed harness invocation can be reused on parent recovery, but an interrupted model/tool turn fails with harness_recovery_required instead of launching a fresh plan automatically. Inspect the recorded tool steps, reconcile changes, and explicitly start a follow-up task or supported session continuation. Different model- generated tool-call IDs represent new operations; the adapter does not deduplicate commands by their text or promise exactly-once external effects.

Routing currently requires a POSIX host and an ATF Harness version supporting trusted extensions, dynamic tool registration, and ExtensionContext.invokeTool. It supports only the six tools shown above, in local mode; other providers and remote transport are refused. All workers serving this node version must share the private journal and workspace on the same host. Keep the journal when restarting workers. Register a new node version when adding these capabilities. The trusted extension disables discovery of unrelated extensions; initialization or routing failures block tools and fail the invocation without local fallback.

Providers

An unknown provider raises ValueError listing supported names. Selection follows call, agent, environment, then default. Providers send prompts after an argument terminator or through stdin so prompt text cannot become a command flag. test_provider_commands_restrict_tools_and_keep_prompts_out_of_flags checks those command builders.

app.harness_doctor() and the module-level helper return ProviderHealth records with provider, binary, availability, path, tool-restriction support and installation hint. They inspect executable paths only. Expose the helper as a skill when operators need to inspect a remote worker's environment:

from __future__ import annotations

from atfgraph import Agent

app = Agent(node_id="delegate", version="1.0.0", description="Harness delegation with typed results")


@app.skill()
def doctor() -> dict:
    """Which harness CLIs are available here (no process is started)."""
    return {health.provider: {"available": health.available, "binary": health.binary} for health in app.harness_doctor()}

A Claude Code node, end to end

This node demonstrates invoking Claude Code and validating its response as a Review:

# review_node/agent.py
from typing import List

from pydantic import BaseModel

from atfgraph import Agent

app = Agent(node_id="reviewer", version="1.0.0", description="Delegates code review to Claude Code")


class Review(BaseModel):
    summary: str
    risks: List[str]
    suggested_tests: List[str]


@app.traversal(max_attempts=1, timeout_seconds=900)  # never retry a harness run
async def review(path: str) -> Review:
    result = await app.harness(
        "Review the code in this directory. Name the main risks and the tests you would add.",
        provider="claude-code",   # the `claude` CLI on the worker's PATH, signed in
        cwd=path,                 # working directory, not a sandbox
        tools=[],                 # all Claude Code built-in tools off: read-only reasoning
        max_turns=6,              # Claude Code option, passed through
        schema=Review,            # the answer must parse as Review
        timeout=600,              # the bridge's wall-clock deadline; timeout_seconds above must be larger
        max_cost_usd=0.50,        # kill the run if Claude Code's streamed cost passes this
    )
    return result.parsed

atfg install ./review_node && atfg run reviewer
atfg traverse reviewer.review --in '{"path": "/Users/you/some-project"}'

The worker starts claude in path with built-in tools disabled and a six-turn limit. With no read tools, this configuration demonstrates the invocation and schema handling; it cannot inspect repository files through those disabled tools. Enable the required read tools for a repository review.

Progress summaries appear on the traversal. The answer must validate as Review. Reported cost is available in result.cost_usd; exceeding a reported-cost cap terminates the child and fails with harness_cost_exceeded. Use doctor to check that claude exists before calling it.

For supervised edits, first obtain a plan with suitable read permissions, pause for request_approval, then run an approved write phase. The supervised-write example shows that structure.

Claude Code uses its own login or ANTHROPIC_API_KEY. tools=[] maps to --tools "". Provider options pass through permission_mode, max_turns, system_prompt and resume. Its cumulative total_cost_usd supports cost checks while events stream.

Delegate with typed output

Pass a Pydantic model or JSON Schema dict as schema. The prompt requests JSON matching that schema. The bridge removes an optional code fence, validates the result and stores it in result.parsed. Invalid output raises HarnessFailure("Harness output did not match the requested schema"). The harness-delegate example is:

from __future__ import annotations

import os
from typing import List, Optional

from pydantic import BaseModel

from atfgraph import Agent

app = Agent(node_id="delegate", version="1.0.0", description="Harness delegation with typed results")


class DirectorySummary(BaseModel):
    purpose: str
    languages: List[str]
    entry_points: List[str]


@app.traversal(timeout_seconds=900)
async def describe_directory(path: str, provider: Optional[str] = None, model: Optional[str] = None) -> DirectorySummary:
    """Ask a coding agent to describe ``path``; the result must match ``DirectorySummary``."""
    if not os.path.isdir(path):
        raise ValueError(f"{path!r} is not a directory")
    result = await app.harness(
        "Describe this directory: its purpose, the programming languages used and the entry points.",
        provider=provider,
        model=model,
        cwd=path,
        schema=DirectorySummary,
        timeout=600,
    )
    return result.parsed

The bridge's timeout=600 and the capability's timeout_seconds=900 control different deadlines. Keep the capability budget larger so the server does not time out while the CLI is still running. The schema-and-doctor test uses fake atfHar to check fenced-JSON parsing and rejection of invalid output.

Working directory and tools

For local mode, provide an existing cwd, either on the call or through Agent(harness_cwd=...). Without one, the SDK raises ValueError("harness requires an explicit cwd") before execution. The process starts in that directory, and providers may also pass --cwd, -C or --dir. This selects location without isolating access to other paths.

Tool handling depends on the provider. ATF Harness and Claude Code disable built-ins without an explicit list; Codex uses read-only mode. tools=["Read", "Grep"] passes individual tool names where supported. A name containing a comma or beginning with - raises ValueError. Providers without equivalent restrictions require explicit acceptance of their defaults, as shown in the provider table.

Timeouts, budgets and cancellation

Cost enforcement depends on reported data. Claude Code emits cumulative cost, OpenCode reports per-step cost, and Codex reports per-turn cost when available. A streamed overrun sends SIGTERM, then SIGKILL after five seconds. HarnessCostExceeded is non-retryable, with code harness_cost_exceeded and fields for reported spend, cap, turns, provider, killed state and partial text. Its message includes the last two KiB of assistant text. Providers without streamed cost are checked at completion when a final price is available. A missing price cannot establish a dollar limit.

This differs from server budgets: budgets can deny later admissions over a window; the harness cap acts on the current process.

Cancelling the traverser cancels the awaiting coroutine. Cleanup sends SIGTERM to the process group, followed by SIGKILL after two seconds if needed. It does not return partial output.

The shipped harness example uses one attempt, disabled tools and a fixed deadline:

from __future__ import annotations

import os
from typing import Optional

from atfgraph import Agent

app = Agent(node_id="harness-example", version="0.1.0", tags=["harness"])


@app.traversal(max_attempts=1, timeout_seconds=180)
async def answer(prompt: str, model: str, cwd: str, provider: Optional[str] = None, timeout: float = 120.0) -> dict:
    if not os.path.isdir(cwd):
        raise ValueError(f"{cwd!r} is not an existing directory")
    result = await app.harness(
        prompt,
        provider=provider,
        model=model,
        cwd=cwd,
        tools=[],
        timeout=float(timeout),
    )
    return {
        "text": result.text,
        "provider": result.provider,
        "model": result.model,
        "input_tokens": result.input_tokens,
        "output_tokens": result.output_tokens,
        "total_tokens": result.total_tokens,
        "cost_usd": result.cost_usd,
        "session_id": result.session_id,
        "turns": result.turns,
        "duration_ms": result.duration_ms,
    }

A harness may change real files, so one attempt avoids automatic replay of edits. This example reads missing model and cwd inputs from ATFGRAPH_HARNESS_MODEL and ATFGRAPH_HARNESS_WORKDIR, and raises ValueError if required values are still absent.

What comes back

HarnessResult includes text (also exposed as result), provider, session ID, model, usage, return code, parsed output, duration, turns, reported USD cost and optional raw events. Usage keys are input, output, cacheRead and cacheWrite, with token-count convenience properties. Prices come only from CLI reports, such as Claude Code's total or OpenCode's step costs. If a provider reports no price, cost remains None; it is not inferred from tokens.

Local Agent.harness does not automatically add usage to app.execution_cost() or the server ledger. Record its returned usage when you want it included:

from __future__ import annotations

from atfgraph import Agent, CostEntry

app = Agent(node_id="delegate", version="1.0.0")


@app.traversal(timeout_seconds=900)
async def summarize(path: str) -> dict:
    result = await app.harness("Summarize this directory in one paragraph.", cwd=path, timeout=600)
    await app.record_usage(
        CostEntry(
            model=result.model or "unknown",
            provider=result.provider,
            source="harness",
            operation="chat",
            input_tokens=result.input_tokens,
            output_tokens=result.output_tokens,
            total_tokens=result.total_tokens,
            cost_usd=result.cost_usd,
            cost_source="provider" if result.cost_usd is not None else None,
        )
    )
    return {"text": result.text, "cost": app.execution_cost()}

CostEntry is exported by atfgraph, and source="harness" identifies this usage category. See Observability for storage and queries.

How a run is recorded on the path

The CLI invocation remains part of its calling traversal; it is not submitted as a child capability. For each JSON event, the bridge durably forwards a summary shaped as {"harness": {"provider": ..., "type": ...}}. Raw event payloads are excluded because tool data may contain secrets. An asynchronous on_event callback receives the same summaries locally.

The fixture-harness integration test runs the real graph daemon and a fake atfHar. It checks harness progress events, returned session/provider/model/token fields, and a HarnessFailure when the executable exits with 1. The example driver prints the event count. This proves bridge-to-graph wiring, rather than a current vendor CLI integration.

atfHar end to end

The default atf-harness provider needs atfHar on the worker's PATH, model configuration under ~/.atf-harness/agent/, and the configured credential in that worker's environment.

1. Configure a model. config.yml selects a role. models.yml maps the model to an endpoint and a credential variable. A Databricks serving configuration looks like this:

# ~/.atf-harness/agent/config.yml
modelRoles:
  default: databricks/claude-sonnet-4-endpoint:max      # provider/model:variant

# ~/.atf-harness/agent/models.yml
providers:
  databricks:
    baseUrl: https://adb-1234567890.12.azuredatabricks.net/serving-endpoints
    apiKey: DATABRICKS_TOKEN        # the NAME of a variable in the worker's environment
    models:
      - id: claude-sonnet-4-endpoint

Not verified: these exact models.yml fields against the current ATF Harness release. The files, provider/model:variant role format and apiKey: <ENV_NAME> resolver follow the original harness guide.

2. Define the node. This shortened example exposes one harness traversal and a doctor skill. It permits no tools and uses one attempt:

from __future__ import annotations

import os
from typing import Optional

from atfgraph import Agent

app = Agent(node_id="harness-example", version="0.1.0", tags=["harness"])


@app.skill()
def doctor() -> dict:
    return {h.provider: {"available": h.available, "path": h.path, "hint": h.hint} for h in app.harness_doctor()}


@app.traversal(max_attempts=1, timeout_seconds=180)
async def answer(prompt: str, model: Optional[str] = None, cwd: Optional[str] = None,
                 variant: Optional[str] = None, timeout: float = 120.0) -> dict:
    model = model or os.environ.get("ATFGRAPH_HARNESS_MODEL") or None
    cwd = cwd or os.environ.get("ATFGRAPH_HARNESS_WORKDIR") or None
    if model is None or cwd is None:
        raise ValueError("set ATFGRAPH_HARNESS_MODEL and ATFGRAPH_HARNESS_WORKDIR, or pass model= and cwd=")
    result = await app.harness(
        prompt,
        provider="atf-harness",
        model=model,
        cwd=cwd,
        tools=[],                 # explicit: no built-in tools
        variant=variant,          # atfHar --thinking; None keeps the configured variant
        timeout=float(timeout),
    )
    return {
        "text": result.text,
        "session_id": result.session_id,
        "model": result.model,
        "turns": result.turns,
        "input_tokens": result.input_tokens,
        "output_tokens": result.output_tokens,
        "cost_usd": result.cost_usd,      # None: atfHar reports no price and none is invented
        "duration_ms": result.duration_ms,
    }

The adapter builds this command. The prompt follows -- so it cannot be parsed as an option:

atfHar --print --mode json --cwd <cwd> --no-title --model <model> [--thinking <variant>] --no-tools --no-session -- <prompt>

An empty tool list selects --no-tools. Omitting session_dir selects --no-session. To allow named tools, pass tools=["read", "glob"]; to retain sessions, provide session_dir.

3. Start the worker and call it. The worker launches atfHar, so set its credential in the worker environment:

export DATABRICKS_TOKEN=dapi-your-token            # the variable models.yml names
export ATFGRAPH_HARNESS_MODEL='databricks/claude-sonnet-4-endpoint'
export ATFGRAPH_HARNESS_WORKDIR="$PWD"
python -m atfgraph.worker --module examples.harness.python.agent:app

# another terminal
python -m examples.harness.python.call --doctor
python -m examples.harness.python.call "Reply with the word ready."
atfg traverse harness-example.answer --in '{"prompt": "List the top-level files here", "variant": "max"}'

CLI events become durable provider/type progress summaries, such as {"harness": {"provider": "atf-harness", "type": "message_end"}}. The driver counts these after printing the result. A missing or expired DATABRICKS_TOKEN can fail the run with the CLI's 401 in error.message. Doctor cannot check that credential because it only locates binaries.

For code migrated from the original example, the outbound worker needs no PORT. Server-enforced max_attempts=1 replaces HarnessConfig(max_retries=0). The example omits max_turns=1 because atfHar --print runs one turn. Failures now raise HarnessFailure, and CLI progress is recorded on the path.

Multi-turn sessions

Each call is one bounded invocation. To continue a local coding session, give the provider a session_dir, retain its returned session_id, and pass that ID as resume next time. Without a directory, ATF Harness disables sessions and rejects resume. The provider tests cover this pattern:

from __future__ import annotations

from typing import Optional

from atfgraph import Agent

app = Agent(node_id="pairing", version="1.0.0", harness_cwd="/srv/repos/checkout", harness_provider="atf-harness")

SESSIONS = "/srv/harness-sessions"   # outside the coding workspace


@app.traversal(max_attempts=1, timeout_seconds=1800)
async def step(instruction: str, session_id: Optional[str] = None) -> dict:
    result = await app.harness(
        instruction,
        session_dir=SESSIONS,
        resume=session_id,
        tools=["Read", "Grep", "Edit"],
        timeout=1200,
    )
    return {"session_id": result.session_id, "text": result.text, "turns": result.turns}

Store the returned session ID in session memory between human turns. ATF Harness and Claude Code support this resume form; Codex refuses it. Set harness_cwd and harness_provider on the Agent for defaults shared by handlers.

Supervised writes: plan, approve, apply

For file writes, separate planning, approval and application. A read-enabled planning call produces a proposal, durable approval waits for a person, and an approved second call applies the changes.

from __future__ import annotations

from typing import List

from pydantic import BaseModel

from atfgraph import Agent

app = Agent(node_id="fixer", version="1.0.0", harness_cwd="/srv/repos/checkout")


class Plan(BaseModel):
    summary: str
    files_to_change: List[str]
    risk: str


class Report(BaseModel):
    files_changed: List[str]
    tests_pass: bool
    summary: str


@app.traversal(max_attempts=1, timeout_seconds=3600)
async def fix_failing_test(test: str) -> dict:
    plan = await app.load_checkpoint("plan")
    if plan is None:
        proposed = await app.harness(f"Plan how to make {test} pass without changing the test. Do not edit anything.",
                                     schema=Plan, tools=["Read", "Grep"], timeout=600)
        plan = proposed.parsed.model_dump()
        await app.save_checkpoint("plan", plan, expected_revision=0)

    decision = await app.request_approval({"question": f"Apply this plan for {test}?", "plan": plan},
                                          operation_key=f"fix:{test}", checkpoint=plan)
    if decision["status"] != "approved":
        return {"status": "rejected", "plan": plan}

    applied = await app.harness(f"Apply this plan and run the tests: {plan['summary']}",
                                schema=Report, tools=["Read", "Grep", "Edit", "Bash"], timeout=1800)
    return {"status": "applied", **applied.parsed.model_dump()}

The checkpoint prevents repeating planning when the handler restarts after approval. One attempt prevents a failed write phase from automatically rerunning against partially edited files. Claude Code receives the listed tool names, such as --tools Read,Grep,Edit. Codex maps enabled tools to workspace-write mode rather than enforcing those names individually. OpenCode requires acceptance of its default tools.

atfHar configuration and Databricks authentication

ATF Harness reads model roles from ~/.atf-harness/agent/config.yml and endpoint and credential resolvers from models.yml. A role uses provider/model:variant. A resolver such as apiKey: DATABRICKS_TOKEN expects that variable in the worker environment. Shells, services and containers have separate environments. A missing variable may be treated as a literal string by the harness and produce a misleading 401.

For a short local Databricks check, obtain a cached OAuth token through an explicitly named profile and pass it to the worker environment:

DATABRICKS_TOKEN="$(databricks auth token <PROFILE> --output json | python3 -c 'import json,sys; print(json.load(sys.stdin)["access_token"])')" \
  python -m atfgraph.worker --module examples.harness.python.agent:app

This token expires, so a long-running worker needs a refresh strategy. The example selects its model with ATFGRAPH_HARNESS_MODEL and variant with ATFGRAPH_HARNESS_THINKING, mapped to --thinking for ATF Harness and model_reasoning_effort for Codex. Keep tokens out of manifests and fixtures. Local harness credentials are not sent to the graph server.

Remote harness: the same call, a workspace on another machine

Remote mode sends the task over HTTPS to an ATF Harness service and executes it in a workspace on that host. ATFGraph still manages the graph run, identity, queueing, cancellation and approval records. The remote harness owns its model loop and tools. This avoids installing a CLI and model credentials on the worker itself.

from atfgraph import Agent
from atfgraph.harness import ExistingWorkspace, RemoteHarness

app = Agent("remote-coder")
remote = RemoteHarness(
    base_url=os.environ["ATFGRAPH_HARNESS_URL"],
    profile="coding-safe",                 # operator-approved profile
    edge_auth="databricks",                # or None: what the proxy in front of the harness wants
    app_token_env="ATFGRAPH_HARNESS_TOKEN",  # optional harness application token, named, never passed
)

@app.traversal(max_attempts=1)
async def implement(task: str) -> dict:
    result = await app.harness(task, transport=remote, workspace=ExistingWorkspace("/workspaces/project"), timeout=1200)
    return {"text": result.text, "session": result.remote.session_id, "turn": result.remote.turn_id}

Choose exactly one execution location: local cwd or remote workspace. Invalid combinations fail before I/O. Remote mode supports only provider="atf-harness". The operator's profile selects tools, model and session behavior, so local options such as tools, model and resume are rejected.

Authentication: The proxy credential uses Authorization. With edge_auth="databricks", the SDK obtains and refreshes it through databricks-sdk unified authentication; install the [databricks] extra for that path. A laptop selects ATFGRAPH_DATABRICKS_PROFILE; an App can use its injected identity. If configured, the separate harness application token is read from its named environment variable and sent in X-Atfhar-Token. Tokens are excluded from reprs, errors and progress summaries.

Sessions and replay: The first call in a graph session creates an idempotent harness session keyed by graph session and workspace, then stores the pairing in session memory. Later calls reuse it. Turn keys derived from the traverser and call let retries and approval continuations reattach to existing turns. result.remote reports session and turn IDs, status, replay indicators, raised request IDs and harness stats.

Approvals: A harness confirmation pauses the traversal using app.request_approval with harness:<session>:<request>. The checkpoint holds references rather than tool payloads. Decide through the Console or Client.decide_approval; continuation answers the harness once and follows its turn. Unsupported interactive input, select or editor requests are cancelled immediately.

Event recovery: The bridge follows SSE events. After overflow, it reconnects from the last sequence. After reset_required, it reads a session snapshot to resynchronize. If streaming ends early, it polls the turn's status.

Readiness and compatibility: app.remote_harness_doctor(remote) checks /healthz, /readyz, /v1/compatibility and profile availability without starting a turn. Local doctor still checks local CLIs. Every remote call checks protocol v1 before creating a session, raising protocol_mismatch or protocol_unknown when compatibility cannot be established.

The remote adapter expects a pre-existing workspace. It does not clone repositories or return patch artifacts. A harness restart during an active turn cannot transparently resume that turn; inspect its terminal status and recover at the application level.

To run examples/harness/remote/, set the harness URL, profile and workspace variables, then use:

python -m atfgraph.worker --module examples.harness.remote.agent:app
python -m examples.harness.remote.call --doctor
python -m examples.harness.remote.call "Fix the failing parser tests and explain the change"
python -m examples.harness.remote.call --session <id> "Now add a regression test"
atfg traverse remote-coder.implement --in '{"task": "..."}' --new-session

Verified by repository tests: remote-protocol behavior against an in-memory harness double and an isolated graph daemon (test_remote_harness_*.py, test_remote_coder_example.py). Not verified by those tests: a live harness deployment or its Databricks Apps proxy. Those require deployment acceptance.

Where a harness cannot run

Local mode requires the executable and its runtime on the worker host. For example, ATF Harness requires Bun. If a hosted image lacks them, install them in a suitable image or use remote mode. Doctor reports available: false for a missing binary, and invocation raises HarnessFailure("<binary> is not installed in this worker environment"). A local cwd always refers to the worker host; it cannot select files on a developer's separate laptop.

Testing a harness capability

Local bridge tests substitute fake CLIs that emit the provider's expected JSON-lines format. This makes parsing, limits and lifecycle tests deterministic without paid provider calls. Use binary= to select the fake executable. The authoring tests build one as follows:

from __future__ import annotations

import sys
from pathlib import Path

import pytest

from atfgraph import Agent


def fake_cli(tmp_path: Path, name: str, code: str) -> str:
    path = tmp_path / name
    path.write_text(f"#!{sys.executable}\n" + code)
    path.chmod(0o755)
    return str(path)


@pytest.mark.asyncio
async def test_describe_directory_parses_the_schema(tmp_path):
    good = fake_cli(tmp_path, "good-atfHar", """import json
print(json.dumps({"type": "message_end", "message": {"role": "assistant",
      "content": '{"purpose": "demo", "languages": ["python"], "entry_points": ["agent.py"]}',
      "stopReason": "stop"}}))
""")
    app = Agent(node_id="delegate", version="1.0.0", harness_cwd=str(tmp_path))
    result = await app.harness("Describe this directory", binary=good, timeout=30)
    assert "purpose" in result.text

The migrated-operations test also runs a fake atfHar worker against a real graph daemon and checks progress and output. Use the daemon fixture for your own end-to-end graph tests.

Run the example

From the repository root, with a server running and ATF Harness installed, use:

export ATFGRAPH_HARNESS_MODEL='your-provider/your-model'
export ATFGRAPH_HARNESS_WORKDIR="$PWD"
python -m atfgraph.worker --module examples.harness.python.agent:app
python -m examples.harness.python.call --doctor
python -m examples.harness.python.call "Reply with the word ready."

The doctor command needs no model or installed CLI; it reports availability from the worker. The task command needs the executable and valid credentials. The examples have no built-in fixture mode. For an offline wiring check, use the fake executable pattern from fake_harness in the migrated-operations tests.

Not verified: these provider adapters against all current real CLI releases. Fake-executable tests, including test_run_harness_with_fake_claude_and_codex_and_opencode, check expected arguments and parsing. Verify an installed version with a live harness-delegate run before relying on that vendor integration.

Next

• Use Model calls for a bounded model/tool loop.
• Require human review with Approvals and checkpoints.
• Record harness spend through Observability.