import asyncio
from sandbox_cli.aio import AsyncStudio
 
async def main():
    studio = await AsyncStudio.connect()
    repo = await studio.project("my-app")
    a, b = await asyncio.gather(
        (await repo.workspace("one")).run(["pytest", "-q"]),
        (await repo.workspace("two")).run(["npm", "test"]),
    )
    print(a.exit_code, b.exit_code)
 
asyncio.run(main())

"""Install a repository's dependencies once, then run its scripts.
 
The everyday shape: a repo with several Python scripts and a requirements.txt,
where the install should happen on the first run and never again.
 
    python3 examples/python_project.py pyqualys                       # setup, then the tests
    python3 examples/python_project.py pyqualys example/example_report.py
 
**The virtualenv lives in the worktree, which is what makes this cheap.** Each
run is a new container and nothing outside `/workspace` survives it — but the
worktree does, so `.venv/` built by one run is there for the next. The setup is
skipped when it is already present, which turns a two-minute first run into a
one-second second one.
 
**The image ships python3 and no pip**, deliberately (see
docs/proposals/python-in-the-image.md), so the setup builds a venv *without* pip
and bootstraps pip inside it. `pip install` into the system interpreter is
refused by PEP 668, correctly, even in a container. Three hosts have to be named
on the egress allowlist for that, and they are named here rather than assumed.
 
**Nothing is installed on your machine.** The dependencies land in a container's
view of a worktree the daemon owns; your interpreter never sees them.
"""
 
from __future__ import annotations
 
import sys
 
from sandbox_cli import Studio
 
# The hosts the install needs. `allow` adds to the daemon's posture and can never
# loosen it — and naming hosts turns the allowlist *on* for a run whose daemon
# had none, so this trades the rest of the internet for these three.
PIP_HOSTS = ["bootstrap.pypa.io", "pypi.org", "files.pythonhosted.org"]
 
# Hosts the *script* needs, as opposed to the install. Empty by default, because
# most scripts need nothing and a run that asks for no egress gets none. Add the
# API your script talks to — e.g. ["qualysapi.qualys.com"] — and nothing else
# becomes reachable by doing so.
SCRIPT_ALLOW: list[str] = []
 
SETUP = [
    ["python3", "-m", "venv", "--without-pip", ".venv"],
    # Into the worktree, not /tmp: the next step is a different container, and
    # only the worktree crosses that boundary.
    ["sh", "-c", "curl -fsSL https://bootstrap.pypa.io/get-pip.py -o get-pip.py"],
    ["sh", "-c", ".venv/bin/python3 get-pip.py -q && rm -f get-pip.py"],
    # Guarded, and with the install's own status preserved: `A && B || true`
    # binds as `(A && B) || true`, so it swallows a *pip* failure as well as an
    # absent file, and the first symptom is an import error much later.
    ["sh", "-c", "if [ -f requirements.txt ]; then .venv/bin/pip install -q -r requirements.txt; fi"],
    # If the repository is itself a package, install it too. Without this, its
    # own scripts fail with `No module named <the repo>` — they import the
    # package they live beside, which only resolves when it is on the path.
    # `|| true` because plenty of repositories are a folder of scripts with no
    # setup.py, and that is not a failure.
    ["sh", "-c", "if [ -f setup.py ] || [ -f pyproject.toml ]; then .venv/bin/pip install -q -e .; fi"],
]
 
 
def main() -> int:
    repo_name = sys.argv[1] if len(sys.argv) > 1 else "my-app"
    script = sys.argv[2] if len(sys.argv) > 2 else None
 
    studio = Studio.connect()
    repo = studio.project(repo_name)
    ws = repo.workspace("deps")
    # A finished run keeps its branch's container name, and this script is a new
    # process every time — so the run it left behind last time is not one it may
    # clear implicitly. This is the line that makes it re-runnable.
    ws.clear_finished()
 
    if not venv_present(ws):
        print("installing dependencies (first run only)…")
        for i, step in enumerate(ws.steps(SETUP, allow=PIP_HOSTS, timeout=900), 1):
            if step.exit_code != 0:
                print(f"setup step {i} failed ({step.exit_code}):\n{step.stderr}", file=sys.stderr)
                # The commonest cause, and worth naming rather than leaving to a
                # traceback: the container could not reach the package index.
                print(f"\nIf that is a network error, the run needs {', '.join(PIP_HOSTS)} "
                      f"on the allowlist.", file=sys.stderr)
                return step.exit_code
        print("  installed")
    else:
        print("dependencies already present in the worktree — skipping setup")
 
    # Whatever the repository already has. The install hosts are *not* passed
    # here: pip is done, and a run that does not ask for egress does not get it.
    #
    # A script that talks to an API needs its host named — SCRIPT_ALLOW below.
    # Running this against pyqualys' example_report.py made a real request and
    # got a 403 back from the vendor's edge, which is the right kind of failure:
    # the code ran, the network was reachable because this daemon allows it, and
    # the only thing missing was a credential. On an allowlist daemon the same
    # script fails earlier and more clearly, at the connection.
    # Not piped through `tail`. POSIX sh has no pipefail, so a pipeline's status
    # is the *last* command's — `tail` always succeeds, and a red test suite
    # would have been reported as exit 0 by an example whose whole job is saying
    # what happened. The output is trimmed here instead, where trimming cannot
    # change a verdict.
    argv = ([".venv/bin/python3", script] if script
            else [".venv/bin/python3", "-m", "unittest", "discover", "-s", ".", "-t", "."])
    out = ws.run(argv, timeout=900, **({"allow": SCRIPT_ALLOW} if SCRIPT_ALLOW else {}))
    # unittest writes its report to stderr, and a script may use either.
    report = (out.stdout.rstrip() + "\n" + out.stderr.rstrip()).strip()
    print(tail(report, 12))
    print(f"exit {out.exit_code}")
    return out.exit_code
 
 
def tail(text: str, lines: int) -> str:
    """The last few lines, trimmed here rather than in the container — where a
    pipe would have replaced the run's exit code with `tail`'s."""
    kept = text.splitlines()[-lines:]
    return "\n".join(kept)
 
 
def venv_present(ws) -> bool:
    """Ask the worktree, not a flag in this process.
 
    The state that matters is on the daemon's disk, and it outlives this script —
    so a variable here would be wrong the moment somebody runs the script twice,
    or deletes the worktree, or another script sets it up first.
    """
    return ws.run(["sh", "-c", "test -x .venv/bin/python3 && echo yes || echo no"]).stdout.strip() == "yes"
 
 
if __name__ == "__main__":
    raise SystemExit(main())
 

"""Serve a repository's FastAPI app in a sandbox, and reach it from here.
 
    python3 examples/fastapi_service.py                 # this repository, auto-detected
    python3 examples/fastapi_service.py my-api app.py   # a registered repo, a named module
 
It runs **the app the repository already has**. An earlier version of this file
cloned a repository, ignored everything in it, wrote an `app.py` on the fly and
served that — which demonstrated nothing about the repository and quietly implied
the clone had done some work. If no app is found, this one scaffolds a tiny app
and *says so*, because a fallback that looks like the real thing is the problem
that version had.
 
Cloning belongs in `python_project.py`, which is about bringing code in and
installing it. This example is about the part that file cannot show:
 
**A server never exits, so `run()` is the wrong verb.** `run()` waits, and
waiting on a server means reaching the deadline and then reporting a container
somebody stopped — a verdict on nothing. `start()` launches and returns the run;
the container outlives this script until something stops it.
 
**`publish` binds the port on the daemon's host**, which is why the health check
at the end can reach it — and why it only works when that host is this one.
 
**Configuration crosses as environment, per workspace.** `read_env_file` parses a
`.env` you name explicitly; `workspace(env=...)` applies it to every run there.
Values travel in the request body, so against a remote daemon without TLS they
cross in cleartext.
 
**The image has python3 and no pip**, deliberately — see
docs/proposals/python-in-the-image.md — so the setup builds a venv *without* pip
and bootstraps pip inside it, which needs three hosts named on the allowlist.
"""
 
from __future__ import annotations
 
import base64
import json
import sys
import time
import urllib.error
import urllib.request
 
from sandbox_cli import Studio
from sandbox_cli.env import read_env_file
 
BRANCH = "fastapi-demo"
PORT = 8123
 
# The hosts the setup needs, named rather than implied. `allow` adds to the
# daemon's posture and can never loosen it — and on a daemon whose egress is
# unrestricted, naming hosts turns the allowlist *on* for that run, so this is
# giving up the rest of the internet rather than asking for more of it.
SETUP_ALLOW = ["bootstrap.pypa.io", "pypi.org", "files.pythonhosted.org"]
 
# Where an app usually lives, in the order worth looking. Checked in the
# worktree, because that is where the run will look for it.
CANDIDATES = ["app.py", "main.py", "api.py", "src/main.py", "app/main.py"]
 
# The setup, once per worktree. The image ships python3 and no pip, so this is a
# venv *without* pip with pip bootstrapped inside it — `pip install` into the
# system interpreter is refused by PEP 668, correctly, even in a container.
SETUP = [
    ["python3", "-m", "venv", "--without-pip", ".venv"],
    # Into the worktree, not /tmp: the next step is a different container, and
    # only the worktree crosses that boundary.
    ["sh", "-c", "curl -fsSL https://bootstrap.pypa.io/get-pip.py -o get-pip.py"],
    ["sh", "-c", ".venv/bin/python3 get-pip.py -q && rm -f get-pip.py"],
    # The repository's own dependencies, when it has any. Serving its app means
    # its imports have to resolve, which is the whole difference between running
    # a repository's code and running code written into a repository.
    # `if`, not `A && B || true`: the latter binds as `(A && B) || true` and
    # swallows a *pip* failure as well as an absent file — after which uvicorn
    # dies on an import and the only symptom is a health check that times out.
    ["sh", "-c", "if [ -f requirements.txt ]; then .venv/bin/pip install -q -r requirements.txt; fi"],
    ["sh", "-c", "if [ -f setup.py ] || [ -f pyproject.toml ]; then .venv/bin/pip install -q -e .; fi"],
    # And the server itself, which the repository may not list because it is not
    # the repository's business how you run it.
    ["sh", "-c", ".venv/bin/pip install -q fastapi uvicorn"],
]
 
DEMO_APP = '''
import os
from fastapi import FastAPI
 
app = FastAPI()
 
@app.get("/health")
def health():
    # Reads configuration that came from the .env on the host, to prove it
    # arrived rather than to do anything useful with it.
    return {"ok": True, "service": os.environ.get("SERVICE_NAME", "unset"),
            "env": os.environ.get("APP_ENV", "unset")}
'''
 
 
def main() -> int:
    repo_name = sys.argv[1] if len(sys.argv) > 1 else None
    named_module = sys.argv[2] if len(sys.argv) > 2 else None
 
    studio = Studio.connect()
    repo = studio.project(repo_name) if repo_name else studio.project()
 
    # Configuration: a file you name, not one that is found for you.
    env = read_env_file(".env.demo", missing_ok=True) or {
        "SERVICE_NAME": "demo-api",
        "APP_ENV": "sandbox",
    }
    ws = repo.workspace(BRANCH, env=env)
    ws.clear_finished()
 
    module = named_module or find_app(ws)
    if module is None:
        # Said plainly rather than done quietly: nothing here is the
        # repository's, so nothing about it is being demonstrated.
        print("no FastAPI app found in this repository — scaffolding demo_app.py "
              f"(looked for: {', '.join(CANDIDATES)})")
        put_demo_app(ws)
        module = "demo_app.py"
    else:
        print(f"serving the repository's own app: {module}")
 
    print("setting up…")
    setup = ws.steps(SETUP, allow=SETUP_ALLOW, timeout=900)
    for i, step in enumerate(setup, 1):
        if step.exit_code != 0:
            print(f"step {i} failed ({step.exit_code}):\n{step.stderr}", file=sys.stderr)
            return step.exit_code
    print(f"  {len(setup)} steps ok")
 
    # `start`, not `run`: this is not meant to finish.
    target = uvicorn_target(module)
    run = ws.start(
        [".venv/bin/uvicorn", target, "--host", "0.0.0.0", "--port", str(PORT)],
        publish=[f"{PORT}:{PORT}"],
    )
    print(f"serving {target} as run {run['id'][:12]} on http://127.0.0.1:{PORT}")
 
    try:
        body = wait_for_health(f"http://127.0.0.1:{PORT}/health", ws, run["id"])
        print("health:", json.dumps(body))
        return 0
    finally:
        # Explicit, because nothing reaps a started run for you. Stopped rather
        # than removed: the logs are the evidence for what it did.
        print("stopping…")
        ws.stop(run["id"])
 
 
def uvicorn_target(module: str) -> str:
    """`app/main.py` -> `app.main:app`, and `app.main:api` -> itself.
 
    Blindly stripping three characters turned `main` into `n:app` and
    `app:app` into `a:app`, and the only symptom either way was a health check
    timing out twenty seconds later — a user typing what uvicorn itself takes
    got the least useful failure available.
    """
    if ":" in module:            # already a uvicorn target
        return module
    path = module[:-3] if module.endswith(".py") else module
    return path.strip("/").replace("/", ".") + ":app"
 
 
def find_app(ws) -> str | None:
    """The first candidate that exists *and* looks like a FastAPI app.
 
    Asked of the worktree rather than of this machine: the repository lives on
    the daemon's disk, and against a remote daemon this script cannot see it at
    all.
    """
    probe = " ; ".join(
        f'test -f {c} && grep -lq "FastAPI(" {c} && echo {c}' for c in CANDIDATES
    )
    found = ws.run(["sh", "-c", f"({probe}) 2>/dev/null | head -1"]).stdout.strip()
    return found or None
 
 
def put_demo_app(ws) -> None:
    """Write the fallback app, base64 so nothing in it can be shell."""
    b64 = base64.b64encode(DEMO_APP.encode()).decode()
    ws.run(["sh", "-c", f"printf %s '{b64}' | base64 -d > demo_app.py"])
 
 
def wait_for_health(url: str, ws, run_id: str, attempts: int = 40) -> dict:
    """Poll until the server answers.
 
    A published port is bound on the **daemon's** host, so this only works when
    that host is this one. On failure it prints the run's own output: a server
    that did not start has a reason, and it is in its logs rather than in the
    timeout.
    """
    last = ""
    for _ in range(attempts):
        try:
            with urllib.request.urlopen(url, timeout=2) as r:
                return json.load(r)
        except (urllib.error.URLError, OSError, ValueError) as e:
            last = str(e)
            time.sleep(0.5)
 
    tail = [l.get("text", "") for l in ws.logs(run_id)][-8:]
    raise RuntimeError(
        f"the server never answered on {url} ({last}).\nIts last output:\n  "
        + "\n  ".join(tail or ["(nothing)"])
    )
 
 
if __name__ == "__main__":
    raise SystemExit(main())
 

"""Fetch a stock price with untrusted code, and let it reach exactly one host.
 
The point of this example is not the price. It is the two lines that decide what
the code can do:
 
    allow=["query1.finance.yahoo.com"]     # the only host it may reach
    ws.run(["python3", "-c", SOURCE])      # no shell, so nothing to quote
 
`allow` **adds** to the daemon's posture and can never loosen it — the same
tighten-only rule a project config gets. What that means in practice is worth
measuring rather than assuming, because it is not "permit these extras":
 
    without allow:  urlopen("https://example.com")  -> 200
    with    allow:  urlopen("https://example.com")  -> blocked
 
…on a daemon whose egress is otherwise **unrestricted**. Naming a host turns the
allowlist *on* for that run, and everything unnamed is then refused. So this
example is not asking for more reach than it had; it is giving up the rest of the
internet in exchange for one host. On a daemon already running an allowlist it
adds one domain, and on `mode: none` it changes nothing and the fetch fails,
which is the correct outcome rather than a surprise.
 
The code travels as an argv element, not through `sh -c`. Nothing parses it, so a
quote or a `$(...)` in the source is just text — the hazard that makes string
interpolation into a shell command a bad habit does not exist here.
"""
 
import json
import sys
 
from sandbox_cli import ApiError, Studio
 
SYMBOL = sys.argv[1] if len(sys.argv) > 1 else "TSLA"
QUOTE_HOST = "query1.finance.yahoo.com"
 
# Runs inside the container. Standard library only: the image ships python 3.11
# and no pip, so anything from PyPI would need a different image or a pip layer.
FETCH = f'''
import json, urllib.request
url = "https://{QUOTE_HOST}/v8/finance/chart/{SYMBOL}?interval=1d&range=1d"
req = urllib.request.Request(url, headers={{"User-Agent": "sandbox-cli-example"}})
with urllib.request.urlopen(req, timeout=20) as r:
    meta = json.load(r)["chart"]["result"][0]["meta"]
print(json.dumps({{
    "symbol": meta.get("symbol"),
    "price": meta.get("regularMarketPrice"),
    "currency": meta.get("currency"),
    "exchange": meta.get("fullExchangeName"),
}}))
'''
 
 
def main() -> int:
    studio = Studio.connect()
    repo = studio.project()               # the repository this script is standing in
    ws = repo.workspace(f"quote-{SYMBOL.lower()}")
    # A finished run keeps its branch's container name, so this is what makes the
    # example runnable twice.
    ws.clear_finished()
 
    try:
        out = ws.run(["python3", "-c", FETCH], allow=[QUOTE_HOST], timeout=90)
    except ApiError as e:
        print(f"the daemon refused the run: {e}", file=sys.stderr)
        return 1
 
    if out.exit_code != 0:
        # The commonest cause by far, and worth naming rather than printing a
        # traceback: the container could not reach the internet.
        print(f"the fetch failed (exit {out.exit_code}):\n{out.stderr}", file=sys.stderr)
        print(
            f"\nIf that is a network error, the daemon's egress posture does not permit "
            f"{QUOTE_HOST}. `allow` can add to what the daemon allows and never loosen it, "
            f"so a daemon started with `mode: none` refuses this by design.",
            file=sys.stderr,
        )
        return out.exit_code
 
    quote = json.loads(out.stdout)
    print(f"{quote['symbol']}  {quote['price']} {quote['currency']}  ({quote['exchange']})")
 
    # The agent variant, for when the question is not "what is the price" but
    # "what should I make of it" — same workspace, same isolation:
    #
    #   verdict = ws.agent("claude", f"Read this quote and say in one line whether "
    #                                f"it moved unusually today: {out.stdout}")
    return 0
 
 
if __name__ == "__main__":
    raise SystemExit(main())
 

"""Three agents that hand work to each other, and a gate that decides.
 
The Python twin of the TypeScript client's `travel-planner.ts`. Async, because
the fan-out is the point: two specialists research in parallel and a coordinator
combines what they produced.
 
Each agent works in its own branch's worktree, which is the isolation unit — one
tree, one container, one agent. So the coordinator **cannot see** what the
specialists wrote, and the artifacts cross deliberately, through this process.
Telling it to "assume the files are there" is the natural thing to write and it
fails silently: the agent invents plausible inputs and the report reads exactly
like one built from the real thing.
 
    python3 examples/travel_planner.py
"""
 
from __future__ import annotations
 
import asyncio
import base64
import sys
from dataclasses import dataclass
 
from sandbox_cli import Outcome, RunCancelled
from sandbox_cli.aio import AsyncStudio, AsyncWorkspace
 
AGENT = "claude"
FALLBACK = ["codex"]
 
TRIP = dict(origin="SFO", destination="NRT", depart="2026-10-15",
            ret="2026-10-22", adults=2, budget_usd=2500)
 
BRIEF = f"""# Trip Brief
- Origin: {TRIP['origin']}
- Destination: {TRIP['destination']}
- Depart: {TRIP['depart']}
- Return: {TRIP['ret']}
- Travelers: {TRIP['adults']} adults
- Rough total budget: ${TRIP['budget_usd']}
"""
 
 
@dataclass
class Finding:
    """What the gate needs to know about one specialist, and nothing else."""
 
    branch: str
    agent: str
    artifact: str
    produced: bool
    note: str
 
 
async def put(ws: AsyncWorkspace, path: str, content: str) -> None:
    """Write a file into a workspace from here.
 
    Base64, not a heredoc. An artifact written by an agent is attacker-controlled
    as far as this script is concerned, and a heredoc built by interpolation is
    one `EOF` line away from being the next command — in a container, as root's
    entrypoint would see it. Base64 has no shell metacharacters, so nothing in
    the content can change what runs.
    """
    b64 = base64.b64encode(content.encode()).decode()
    r = await ws.run(["sh", "-c", f"printf %s '{b64}' | base64 -d > {path}"])
    if r.exit_code != 0:
        raise RuntimeError(f"writing {path}: {r.stderr.strip()}")
 
 
async def get(ws: AsyncWorkspace, path: str) -> str:
    """Read a file back out. Empty when it is not there.
 
    Base64 on the way back too, and not for symmetry: `stdout` is the run's log
    *lines* joined, so a file's trailing newline cannot survive `cat` — measured,
    64 bytes back for 65 written. Fine for reading output, wrong for moving a
    file, and the missing byte is one no "did it work" check would notice.
    """
    r = await ws.run(["sh", "-c", f"base64 < {path} 2>/dev/null | tr -d '\\n' || true"])
    raw = r.stdout.strip()
    return base64.b64decode(raw).decode() if raw else ""
 
 
async def specialist(repo, branch: str, artifact: str, prompt: str) -> tuple[AsyncWorkspace, Finding]:
    ws = await repo.workspace(branch)
    # A finished run keeps its branch's container name until something reaps it,
    # so without this the second run of this script is refused.
    await ws.clear_finished()
    await put(ws, "trip-brief.md", BRIEF)
 
    try:
        out: Outcome = await ws.agent(AGENT, prompt, fallback=FALLBACK, timeout=12 * 60)
    except RunCancelled as cancelled:
        # The wait was cancelled and the SDK stopped the run before raising. The
        # container existed, so silence here would leave an agent working with
        # nobody reading the result.
        return ws, Finding(branch, AGENT, artifact, False, f"cancelled ({cancelled.run['id'][:12]})")
 
    if out.stopped:
        # Not a verdict. A container somebody interrupted has no opinion about the
        # work, and reporting it as failure is how a deadline becomes a bug report.
        return ws, Finding(branch, out.agent or AGENT, artifact, False, "outlived its deadline")
    if out.exit_code != 0:
        last = out.stderr.strip().splitlines()
        return ws, Finding(branch, out.agent or AGENT, artifact, False,
                           last[-1] if last else f"exit {out.exit_code}")
 
    # Asked of the filesystem rather than of the agent: an agent that reports
    # success having written nothing is the commonest failure this gate exists to
    # catch, and the one it cannot be told about.
    produced = (await get(ws, artifact)).strip() != ""
    return ws, Finding(branch, out.agent or AGENT, artifact, produced,
                       "ok" if produced else f"finished without writing {artifact}")
 
 
FLIGHTS = f"""You are the flight search specialist. Read trip-brief.md.
 
Research realistic options from {TRIP['origin']} to {TRIP['destination']} for those dates,
preferring nonstop or one stop. Write flights.json:
 
{{"options": [{{"id": "F1", "airline": "...", "price_usd": 850, "stops": 0}}], "recommended": "F1"}}
 
Write the file and stop."""
 
HOTELS = """You are the hotel search specialist. Read trip-brief.md.
 
Find three or four mid-range hotels in Tokyo near transit, $120-250 a night.
Write hotels.json:
 
{"options": [{"id": "H1", "name": "...", "price_per_night_usd": 180}], "recommended": "H1"}
 
Write the file and stop."""
 
 
async def main() -> int:
    studio = await AsyncStudio.connect()
    # A lookup, not a registration: `add_project` would permanently add this
    # directory to the daemon's registry as a side effect of running an example,
    # and against a remote daemon it would post a local path that daemon cannot
    # resolve.
    repo = await studio.project()          # the repository this script is standing in
 
    # In parallel, because the isolation unit is the branch: two agents in one
    # tree would be a data race with a filesystem in the middle; two agents in
    # two trees are simply two runs. return_exceptions rather than bare gather —
    # one specialist failing is not a reason to lose the other's work, and the
    # coordinator is told what is missing instead.
    outcomes = await asyncio.gather(
        specialist(repo, "agent-flights", "flights.json", FLIGHTS),
        specialist(repo, "agent-hotels", "hotels.json", HOTELS),
        return_exceptions=True,
    )
 
    sources: list[tuple[AsyncWorkspace, Finding]] = []
    findings: list[Finding] = []
    for branch, artifact, result in (("agent-flights", "flights.json", outcomes[0]),
                                     ("agent-hotels", "hotels.json", outcomes[1])):
        if isinstance(result, BaseException):
            findings.append(Finding(branch, "?", artifact, False, repr(result)))
            continue
        sources.append(result)
        findings.append(result[1])
 
    for f in findings:
        print(f"{'OK  ' if f.produced else 'SKIP'} {f.branch:<15} {f.agent:<7} {f.note}")
 
    # The handover: read each artifact out of the tree that produced it and write
    # it into the coordinator's. This is the step that makes this a workflow
    # rather than three agents guessing in parallel.
    coord = await repo.workspace("agent-coordinator")
    await coord.clear_finished()
    await put(coord, "trip-brief.md", BRIEF)
 
    handed: list[str] = []
    for ws, f in sources:
        if not f.produced:
            continue
        await put(coord, f.artifact, await get(ws, f.artifact))
        handed.append(f.artifact)
    print(f"handed over: {', '.join(handed) or 'nothing'}")
 
    task = ("Pick the best flight and hotel combination that stays near the budget."
            if len(handed) == 2 else
            "Some specialist output is missing. Say so explicitly in the itinerary and work "
            "with what is here — do not invent the missing file's contents.")
 
    final = await coord.agent(
        AGENT,
        f"You are the travel coordinator. You have trip-brief.md"
        f"{' and ' + ' and '.join(handed) if handed else ''}.\n\n{task}\n\n"
        "Write itinerary.md for a human, and recommendation.json with the final choice and "
        "estimated total. Any booking step is SIMULATED — do not attempt a real payment.",
        fallback=FALLBACK, timeout=10 * 60,
    )
 
    print(f"\ncoordinator exited {final.exit_code}")
    print((await get(coord, "itinerary.md"))[:2000])
 
    # The decision, and the reason it asks two questions: the coordinator's exit
    # code says whether it finished, and the file says whether it decided
    # anything. Either alone has been enough to wave through an empty report.
    decided = (await get(coord, "recommendation.json")).strip() != ""
    if final.exit_code == 0 and decided:
        return 0
    if not decided:
        print("no recommendation was produced", file=sys.stderr)
    return 1
 
 
if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))