diff --git a/README.md b/README.md index 6b95844..25e2107 100644 --- a/README.md +++ b/README.md @@ -230,7 +230,7 @@ GET /api/events paged by row id, not timestamp GET /api/summary enough for a status line GET /api/control is the fleet claiming work? POST /api/control pause, drain or resume -GET /api/roles where each role's calls go +GET /api/roles where each role's calls go, and which are called PUT /api/roles re-route a role, live GET /healthz open, cheap, needs no credential ``` diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index 07c048c..dae1b90 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -184,9 +184,16 @@ misconfigured one it tells you less than the line above. | `checkout` | The project's `work_dir` is missing or is not a git repository *inside this process's filesystem* — a container needs it mounted. | | `disk space` | The volume holding `work_dir` is below the project's configured `min_free_disk_gb` floor. Free and total GiB are included in the detail. | | `github write` | `gh` is missing, unauthenticated, or the account lacks push on that repo. | -| `reviewer` | No reviewer route. `PUT /api/roles`, or restart with `--reviewer`/`--endpoint`. | +| `reviewer` | No reviewer route, globally or on the project. `PUT /api/roles`, restart with `--reviewer`/`--endpoint`, or give the project its own `roles.reviewer`. | +| `role reachability` | The route exists and the model does not answer it. The detail names the model and the status — an endpoint can advertise a model in `/models` and serve nothing behind it. Route the role somewhere that replies. | | `base checks` | A configured command failed on an unmodified base-branch worktree — fix the command or its prerequisites before starting. Also reported when no run has happened yet (`not_run`) or one is still going. | +A project's `roles` override the global map **per role**: a project that names +only a reviewer still inherits every other role, and the reviewer it names is +the one its workers call. + Warnings do not block a start and are still worth reading: `checks` means -nothing verifies a diff before the reviewer sees it, and `reviewer -independence` means some share of reviews is a model grading its own work. +nothing verifies a diff before the reviewer sees it, `reviewer independence` +means some share of reviews is a model grading its own work, and `model +latency` means a model answered a one-token prompt slowly — usable, and every +call pays that first. diff --git a/docs/INTERNALS.md b/docs/INTERNALS.md index d8d8b1a..112ec81 100644 --- a/docs/INTERNALS.md +++ b/docs/INTERNALS.md @@ -372,6 +372,12 @@ This was previously documented in three places and enforced in none. It is deliberate choice, and blocking it would be the harness overruling an operator about their own budget. What it must not be is a surprise. +It compares the reviewer to the implementer that **actually runs**. In session +mode that is the agent process, not a routed model, so the verdict says so +rather than comparing two routes that never meet: the configured `implementer` +is never called there, and a warning about it would be about a pairing that +does not exist. + --- ## 7. Merge, and the only honest quality metric diff --git a/docs/USAGE.md b/docs/USAGE.md index ccefb13..adfd221 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -400,6 +400,13 @@ Takes effect on the next call, no restart. This is possible only because a call site names a **role**, never a model — so re-routing one is a data change rather than a code change. +The response says which of those roles this deployment actually calls. In +session mode the agent process plans and implements with its own credentials +and endpoint, so `planner` and `implementer` come back `"used": false` with the +command that does that work instead: they are stored, the non-session executor +uses them, and nothing here will. A project can override any role for itself +with `roles` on its registration; unnamed roles still come from this map. + Worth doing deliberately: a reviewer on the same vendor as the implementer means some share of reviews is a model grading its own work. Nothing enforces that; it is your call. @@ -471,7 +478,7 @@ GET /api/events paged by row id, not timestamp GET /api/summary enough for a status line GET /api/control is the fleet claiming work? POST /api/control pause, drain or resume — never interrupts work -GET /api/roles where each role's calls go +GET /api/roles where each role's calls go, and which are called PUT /api/roles re-route a role, live GET /api/readiness can anything actually run, and why not GET /healthz open, cheap, needs no credential diff --git a/src/agent_harness/__main__.py b/src/agent_harness/__main__.py index 8ebd9d1..98a8d39 100644 --- a/src/agent_harness/__main__.py +++ b/src/agent_harness/__main__.py @@ -138,8 +138,14 @@ def _http_transport(api_key: str) -> Any: client = httpx.Client(timeout=httpx.Timeout(600.0, connect=30.0)) def transport(route: Any, messages: Any, options: Any) -> Any: + asked = float(options.get("timeout") or 0.0) payload = {"model": route.model, "messages": list(messages)} - payload.update({k: v for k, v in options.items() if k != "role"}) + # `timeout` instructs the transport; it is not a completion parameter, + # and sending it as one would have the provider reject the request. + # Preflight's reachability probe sets it, because a probe that + # inherited the work timeout would take ten minutes to establish that + # a model is not answering. + payload.update({k: v for k, v in options.items() if k not in ("role", "timeout")}) response = client.post( f"{route.endpoint.rstrip('/')}/chat/completions", headers={ @@ -147,6 +153,11 @@ def transport(route: Any, messages: Any, options: Any) -> Any: "content-type": "application/json", }, json=payload, + timeout=( + httpx.Timeout(asked, connect=min(30.0, asked)) + if asked + else httpx.USE_CLIENT_DEFAULT + ), ) return Response(response.status_code, dict(response.headers), response.text) @@ -159,10 +170,9 @@ def _run(args: argparse.Namespace) -> int: import json as _json import shlex - from . import providers from .executor import Checks, Executor from .github import GitHub - from .model_client import ModelClient, Route + from .model_client import ModelClient, Route, routes_from_map from .work import RUNNING, WorkQueue, WorkRecord # With a session host the CLI agent does the implementing, so only the @@ -282,16 +292,7 @@ def emit(event: dict[str, Any]) -> None: print(f"note: the stored role map had no route for {', '.join(filled)}; used the flags.") def live_routes() -> dict[str, Route]: - stored = queue.get_setting(ROLE_MAP_KEY) or {} - return { - name: Route( - route["model"], - route["endpoint"], - providers.PROVIDERS.get(route.get("provider", ""), providers.CLAW_BAY), - api_key=api_key, - ) - for name, route in stored.items() - } + return routes_from_map(queue.get_setting(ROLE_MAP_KEY) or {}, api_key=api_key) # Nothing claims work until every role this run needs can be routed. The # alternative is finding out on the first model call -- after the project @@ -322,7 +323,13 @@ def live_routes() -> dict[str, Route]: # enforced in none, so a reviewer could be the same model as the # implementer and nothing would mention it -- every review a model # grading its own work, invisibly. - independent, why = client.reviewer_independence() + # + # Against the implementer that actually runs: in session mode the agent + # process writes the code, so comparing the reviewer to the configured + # implementer route would be a verdict about a pairing that never happens. + independent, why = client.reviewer_independence( + implemented_by=args.agent if session_mode else "" + ) print(("reviewer: " if independent else "WARNING: ") + why) executor: Any @@ -683,7 +690,9 @@ def main(argv: list[str] | None = None) -> int: ) maintenance.start() - fleet, reviewer_client, host = _fleet_for_serve(args, queue_for_serve, audit=audit) + fleet, reviewer_client, host, executor_roles = _fleet_for_serve( + args, queue_for_serve, audit=audit + ) if fleet is None: print( "monitoring only: no --session-host, so no worker pool is attached and " @@ -704,6 +713,7 @@ def main(argv: list[str] | None = None) -> int: # Readiness probes it with a read. Passing the client rather # than the URL keeps the token out of the API layer. session_host=host, + executor_roles=executor_roles, ), host=args.host, port=args.port, @@ -720,10 +730,10 @@ def main(argv: list[str] | None = None) -> int: def _fleet_for_serve( args: argparse.Namespace, queue: Any, *, audit: Any | None = None -) -> tuple[Any | None, Any | None, Any | None]: +) -> tuple[Any | None, Any | None, Any | None, Any | None]: """The supervised half of `serve`: a fleet the API's start action can use. - Returns (None, None) for a monitoring-only deployment. That mode is + Returns all-None for a monitoring-only deployment. That mode is supported on purpose — a dashboard over someone else's harness should not need a session host, a model key or a checkout — and the API already refuses to start a project when nothing can claim. @@ -732,18 +742,17 @@ def _fleet_for_serve( the API's start action does, and only after preflight passes. """ if not args.session_host: - return (None, None, None) + return (None, None, None, None) import json as _json import shlex - from . import providers from .api import ROLE_MAP_KEY from .events import KINDS, MODEL_CALL, Event from .fleet import Fleet from .github import GitHub - from .model_client import ModelClient, Route - from .runtime import session_executor_factory + from .model_client import ModelClient, Route, effective_routes, routes_from_map + from .runtime import ExecutorRoles, session_executor_factory from .session_executor import AgentSpec from .session_host import HttpSessionHost @@ -766,25 +775,29 @@ def _fleet_for_serve( queue.set_setting(ROLE_MAP_KEY, stored) def live_routes() -> dict[str, Route]: - current = queue.get_setting(ROLE_MAP_KEY) or {} - return { - name: Route( - route["model"], - route["endpoint"], - providers.PROVIDERS.get(route.get("provider", ""), providers.CLAW_BAY), - api_key=api_key, - ) - for name, route in current.items() - if route.get("model") and route.get("endpoint") - } + return routes_from_map(queue.get_setting(ROLE_MAP_KEY) or {}, api_key=api_key) + + def routes_for(project_id: str) -> dict[str, Route]: + """One project's effective map, read live on every call. + + The project row is read here rather than closed over so that a role + override written through the API reaches a worker that is already + running — the same reason the global map is read per call. + """ + project = queue.get_project(project_id) + return effective_routes( + live_routes(), + routes_from_map(getattr(project, "roles", None) or {}, api_key=api_key), + ) routes = live_routes() if "reviewer" not in routes: # Not fatal, and not silent: preflight blocks the start with exactly # this reason, so the fleet may as well exist and say why now. print( - "warning: no reviewer is routed. Preflight will refuse to start any " - "project — set one with --reviewer/--endpoint or PUT /api/roles.", + "warning: no reviewer is routed globally. Preflight will refuse to start " + "any project that does not override one — set a global reviewer with " + "--reviewer/--endpoint or PUT /api/roles.", file=sys.stderr, ) events_path = args.events or Path(args.db).with_name("events.jsonl") @@ -849,11 +862,13 @@ def emit(event: dict[str, Any]) -> None: ) host = HttpSessionHost(args.session_host, token=host_token) + agent = AgentSpec(command=tuple(shlex.split(args.agent))) factory = session_executor_factory( queue, host=host, - agent=AgentSpec(command=tuple(shlex.split(args.agent))), + agent=agent, reviewer=reviewer_client, + routes_for=routes_for, github_for=GitHub, ui_base_url=args.session_host, on_event=emit, @@ -863,7 +878,14 @@ def emit(event: dict[str, Any]) -> None: print(f"events: {events_path}") # The fleet emits into the same stream as the executors: a worker that # dies is recorded next to the work it was doing, not in a separate log. - return (Fleet(queue, factory, poll_seconds=args.poll, on_event=emit), reviewer_client, host) + return ( + Fleet(queue, factory, poll_seconds=args.poll, on_event=emit), + reviewer_client, + host, + # What this deployment will actually call, so the API can stop + # advertising the two roles the agent process does instead. + ExecutorRoles.for_session(agent), + ) if __name__ == "__main__": diff --git a/src/agent_harness/api.py b/src/agent_harness/api.py index ddb76d5..50139a6 100644 --- a/src/agent_harness/api.py +++ b/src/agent_harness/api.py @@ -85,7 +85,9 @@ ResolveQuestion, RetryResult, RoleMap, + RoleMapView, RoleRoute, + RoutedRole, ScopeRequest, SetFleetControl, StopProjectRequest, @@ -114,6 +116,14 @@ #: because the API and the worker are different processes. ROLE_MAP_KEY = "role_map" +#: How long a healthy model has to answer preflight's one-token probe, and +#: how long it has before it is treated as not answering at all. Anything in +#: between is reported as slow and not refused — a late model is usable, and +#: the harness has no business deciding otherwise about someone else's +#: endpoint. +MODEL_PROBE_TIMEOUT = 10.0 +MODEL_PROBE_PATIENCE = 20.0 + DESCRIPTION = """\ Plans work, claims it, runs it as an agent in a terminal session, and records what happened. @@ -156,6 +166,7 @@ def create_api( model_client: Any | None = None, session_host: Any | None = None, probes: Mapping[str, Any] | None = None, + executor_roles: Any | None = None, ) -> FastAPI: """Build the API. @@ -169,6 +180,12 @@ def create_api( every probe injected precisely so a readiness gate could be tested; this layer used to hardcode the defaults, which put a real `gh` subprocess and a real filesystem read behind an HTTP route. + + `executor_roles` says which roles the executor this deployment builds can + actually reach (`runtime.ExecutorRoles`). Without it the role map + advertises every configured route as live, which in session mode was + wrong for two of three stages: the agent process plans and implements, + and no `planner` or `implementer` route is ever called. """ app = FastAPI( title="agent-harness", @@ -188,8 +205,11 @@ def create_api( app.state.session_host = session_host # Everything preflight needs lives on THIS app. It used to be copied into # a module-level dictionary, so a second `create_api` in the same process - # silently took over the first's wiring. + # silently took over the first's wiring -- including, once roles were + # routed per project, which roles it believed its executor could reach. app.state.probes = dict(probes or {}) + app.state.executor_roles = executor_roles + app.state.ask_model = _model_asker(model_client) app.state.base_checks = BaseChecks() app.state.token = token @@ -343,7 +363,7 @@ def retry( detail=f"{item_id} is claimed by {record.owner} and its lease is live; " "wait for the lease to expire rather than racing it", ) - queue.requeue(item_id, project_id=project_id) + queue.release(item_id, PENDING, error=None, project_id=project_id) return RetryResult(ok=True, item_id=item_id, state="pending") @app.post( @@ -867,16 +887,6 @@ def create_project( It starts **stopped**. Registering a project must not begin spending money on it, and nothing here starts a worker -- only an explicit start does. - - Updating a project that is **already running** reconciles its pool to - the new `max_workers`: extra workers are started, surplus ones stop - claiming and are joined once their in-flight item finishes, and no - session is interrupted. Persisting a number that only took effect at - the next start meant every capacity change cost a stop/start cycle -- - lifecycle risk taken on a project whose agents are working, to apply - an integer. `workers` in the response is what is alive now, which - after a shrink is still the old count until those items reach a - boundary; `project.max_workers` is what was asked for. """ queue = need_queue() queue.add_project( @@ -1075,6 +1085,11 @@ def readiness( item is claimed, no control state changes, and no credential is echoed — the session host is probed with a read, which proves both reachability and that the token is accepted. + + It is not free: each routed model is asked for a one-token completion, + because a model that is configured and does not answer is the failure + this is for. The answer is remembered for a minute, so polling this + does not become a load generator against the endpoint. """ queue = need_queue() fleet_ = app.state.fleet @@ -1107,26 +1122,48 @@ def readiness( session_host_state = ReadinessProbe(configured=True, ok=ok, detail=detail) probe = lambda ok=ok, detail=detail: (ok, detail) # noqa: E731 - stored = queue.get_setting(ROLE_MAP_KEY) or {} - route = stored.get("reviewer") or {} - client = app.state.model_client - independence = client.reviewer_independence() if client is not None else None - reviewer_state = ReadinessProbe( - configured=bool(route.get("model")), - ok=bool(route.get("model")), - detail=( - f"reviewer routed to {route['model']}" - + (f"; {independence[1]}" if independence else "") - if route.get("model") - else "no reviewer is routed; every review fails closed, so every item " - "would fail after the implementation was paid for" - ), - ) - projects = [p for p in queue.projects() if project_id is None or p.project_id == project_id] if project_id is not None and not projects: raise HTTPException(status_code=404, detail=f"no project {project_id!r}") + # Global first, then per project — because a project may override the + # reviewer. Reading only the global route made this line contradict + # the project reports underneath it: it announced that nothing could + # be reviewed while every project routed a reviewer of its own. + from .model_client import reviewer_independence + + global_routes = _role_routes(queue) + global_route = global_routes.get("reviewer") + overriding = sorted( + p.project_id + for p in projects + if (p.roles or {}).get("reviewer") and _role_routes(queue, p).get("reviewer") + ) + # Not "some project can review": a reviewer nothing has is a reviewer + # the projects without one still fail closed on. + covered = global_route is not None or (bool(projects) and len(overriding) == len(projects)) + if global_route is not None: + note = reviewer_independence( + global_routes, implemented_by=_executor_roles(app.state).implemented_by + )[1] + reviewer_detail = f"reviewer routed to {global_route.model}; {note}" + if overriding: + reviewer_detail += f"; overridden by project(s): {', '.join(overriding)}" + elif covered: + reviewer_detail = ( + f"no global reviewer; every project routes its own: {', '.join(overriding)}" + ) + else: + reviewer_detail = ( + "no reviewer is routed; every review fails closed, so every item " + "would fail after the implementation was paid for" + ) + reviewer_state = ReadinessProbe( + configured=global_route is not None or bool(overriding), + ok=covered, + detail=reviewer_detail, + ) + reports = [] for project in projects: report = _preflight( @@ -1208,20 +1245,26 @@ def set_control( @app.get( "/api/roles", tags=["control"], - summary="Where each role's calls go", - response_model=RoleMap, + summary="Where each role's calls go, and which of them run", + response_model=RoleMapView, ) - def get_roles(_: None = Depends(require_token)) -> RoleMap: - stored = need_queue().get_setting(ROLE_MAP_KEY) or {} - return RoleMap(roles={name: RoleRoute(**route) for name, route in stored.items()}) + def get_roles(_: None = Depends(require_token)) -> RoleMapView: + """The map, annotated with what this deployment actually calls. + + A route nothing calls is not a harmless extra: this endpoint is how an + operator answers "what am I paying for, and what is grading it?", and + in session mode two of its three answers described a model that is + never asked anything. + """ + return _role_map_view(app.state, need_queue()) @app.put( "/api/roles", tags=["control"], summary="Change the role map without a redeploy", - response_model=RoleMap, + response_model=RoleMapView, ) - def set_roles(request: RoleMap, _: None = Depends(require_token)) -> RoleMap: + def set_roles(request: RoleMap, _: None = Depends(require_token)) -> RoleMapView: """Takes effect on the next model call. This is possible only because a call site names a **role**, never a @@ -1230,6 +1273,11 @@ def set_roles(request: RoleMap, _: None = Depends(require_token)) -> RoleMap: cheaper tier, or the reviewer to a different vendor, while the fleet is running. + The response says which of the roles just stored are actually called. + Storing one that is not still succeeds — the non-session executor uses + it, and so may a later deployment — but it changes nothing about what + runs here, and echoing it back unqualified said otherwise. + A reviewer on the same vendor as the implementer means some share of reviews is a model grading its own work. Nothing here enforces that; it is your call, and it is worth making deliberately. @@ -1237,9 +1285,15 @@ def set_roles(request: RoleMap, _: None = Depends(require_token)) -> RoleMap: queue = need_queue() queue.set_setting( ROLE_MAP_KEY, - {name: route.model_dump() for name, route in request.roles.items()}, + # Only the routing fields. `used` is computed from the deployment, + # not configured, and storing it would let a stale answer be read + # back later as though an operator had set it. + { + name: route.model_dump(include={"model", "endpoint", "provider"}) + for name, route in request.roles.items() + }, ) - return request + return _role_map_view(app.state, queue) # ---------------------------------------------------------------- plan @@ -1498,6 +1552,76 @@ def _audit_event_fields(row: dict[str, Any]) -> dict[str, Any]: } +def _role_routes(queue: WorkQueue, project: Project | None = None) -> dict[str, Any]: + """The routes a project will actually use. + + The global map, with that project's own overrides applied **per role**. + Every reader goes through here because they used to disagree: preflight + took the project map *or* the global one and never merged them, while + readiness read only the global one — so a project could pass preflight on + its own reviewer override and then be executed against a different model, + or refused for the absence of a global route it does not need. + + No API key is attached: these routes are for reading and reporting, and + the transport supplies the credential when a call is actually made. + """ + from .model_client import effective_routes, routes_from_map + + return effective_routes( + routes_from_map(queue.get_setting(ROLE_MAP_KEY) or {}), + routes_from_map(project.roles or {}) if project is not None else {}, + ) + + +def _executor_roles(state: Any) -> Any: + """What the attached executor can reach. Everything, unless told otherwise.""" + from .runtime import ExecutorRoles + + roles = getattr(state, "executor_roles", None) + return roles if roles is not None else ExecutorRoles() + + +def _model_asker(client: Any) -> Any: + """The thing preflight uses to ask a model whether it is there. + + Remembered briefly: readiness is polled, and a probe on every poll would + make a dashboard a load generator against the model endpoint. + """ + if client is None or not hasattr(client, "answers"): + return None + from .preflight import Answer, remembered + + def ask(route: Any) -> Any: + started = time.time() + ok, detail = client.answers(route, timeout=MODEL_PROBE_PATIENCE) + return Answer(ok=ok, detail=detail, seconds=time.time() - started) + + return remembered(ask) + + +def _role_map_view(state: Any, queue: WorkQueue) -> RoleMapView: + """The global map, annotated with what this deployment will call.""" + from .model_client import reviewer_independence + + stored = queue.get_setting(ROLE_MAP_KEY) or {} + executor = _executor_roles(state) + independent, why = reviewer_independence( + _role_routes(queue), implemented_by=executor.implemented_by + ) + return RoleMapView( + reviewer_independent=independent, + reviewer_note=why, + roles={ + name: RoutedRole( + **RoleRoute(**route).model_dump(), + used=executor.calls_role(name), + unused_reason=executor.unused_reason(name), + ) + for name, route in stored.items() + }, + ) + + def _preflight( state: Any, queue: WorkQueue, @@ -1508,35 +1632,52 @@ def _preflight( ) -> Any: """Build a preflight report for a project, using whatever is configured. - `state` is the calling app's own `app.state`. It used to be a module-level - dictionary, which meant a second `create_api` silently overwrote the - first's wiring: app A would report app B's worker pool, session host and - probe results, while its top-level readiness fields still came from its - own state. A readiness gate that can be changed by an unrelated app in the - same process is not a gate. - `session_host` is a *probe*, not the host: readiness over many projects would otherwise ask the same host the same question once per project. """ - from .preflight import last_base_result_probe, preflight_project, session_host_probe + from .model_client import reviewer_independence + from .preflight import ( + last_base_result_probe, + preflight_project, + role_reachability_probe, + session_host_probe, + ) - stored = queue.get_setting("role_map") or {} - client = getattr(state, "model_client", None) + routes = _role_routes(queue, project) + executor = _executor_roles(state) + ask = getattr(state, "ask_model", None) host = getattr(state, "session_host", None) - base_checks = getattr(state, "base_checks", None) - return preflight_project( - project, - has_fleet=getattr(state, "fleet", None) is not None, - reviewer_route=(project.roles or stored).get("reviewer"), - reviewer_independent=client.reviewer_independence() if client is not None else None, - session_host=session_host or (session_host_probe(host) if host is not None else None), - checks_probe=( - last_base_result_probe(base_checks, project.project_id) - if check_base and base_checks is not None + # Only the roles this executor calls. Probing a route nothing will ever + # use spends tokens to answer a question about it, and could refuse a + # start over a model no item depends on. + reachable = {name: route for name, route in routes.items() if executor.calls_role(name)} + kwargs: dict[str, Any] = { + "has_fleet": getattr(state, "fleet", None) is not None, + "reviewer_route": routes.get("reviewer"), + "reviewer_independent": reviewer_independence( + routes, implemented_by=executor.implemented_by + ), + "role_probe": ( + role_reachability_probe( + reachable, + ask, + timeout=MODEL_PROBE_TIMEOUT, + patience=MODEL_PROBE_PATIENCE, + ) + if ask is not None and reachable else None ), - **(getattr(state, "probes", None) or {}), - ) + "session_host": session_host or (session_host_probe(host) if host is not None else None), + "checks_probe": ( + last_base_result_probe(state.base_checks, project.project_id) + if check_base and getattr(state, "base_checks", None) is not None + else None + ), + } + # Injected probes win, so a test can answer any of these without a + # network, a subprocess or a model. + kwargs.update(getattr(state, "probes", None) or {}) + return preflight_project(project, **kwargs) def _project_spec(project: Project) -> ProjectSpec: diff --git a/src/agent_harness/model_client.py b/src/agent_harness/model_client.py index c988c70..a9e08ae 100644 --- a/src/agent_harness/model_client.py +++ b/src/agent_harness/model_client.py @@ -202,6 +202,101 @@ class Response: body: bytes | str +def routes_from_map( + stored: Mapping[str, Mapping[str, Any]] | None, + *, + api_key: str | None = None, + default_provider: Provider = P.CLAW_BAY, +) -> dict[str, Route]: + """The persisted role map, as routes. + + One conversion, shared by `run`, by `serve` and by every readiness + report, so the map an operator reads is the map the fleet calls. A role + missing a model or an endpoint is dropped rather than half-built: it is + not a route, and preflight's job is to name it as missing rather than to + fail on the first call that uses it. + """ + routes: dict[str, Route] = {} + for name, spec in (stored or {}).items(): + model, endpoint = spec.get("model"), spec.get("endpoint") + if not model or not endpoint: + continue + routes[name] = Route( + str(model), + str(endpoint), + P.PROVIDERS.get(str(spec.get("provider", "")), default_provider), + api_key=api_key, + ) + return routes + + +def effective_routes( + global_routes: Mapping[str, Route], project_routes: Mapping[str, Route] | None +) -> dict[str, Route]: + """The global role map with one project's overrides applied. + + Per role, not wholesale. Choosing one map or the other was the defect: + a project that overrode only its reviewer passed preflight on that + override and was then executed against the *global* reviewer, or failed + with `no route for role reviewer` when the global map had none. A partial + project map inherits every role it does not name. + """ + return {**global_routes, **(project_routes or {})} + + +def reviewer_independence( + roles: Mapping[str, Route], *, implemented_by: str = "" +) -> tuple[bool, str]: + """Whether the reviewer is independent of whatever wrote the code. + + Returns (independent, why). This was documented in three places and + enforced in none, which meant a reviewer could be the same model as the + implementer and nothing would say so -- some share of reviews being a + model grading its own work, invisibly. + + `implemented_by` names the thing that actually implements when it is not + a routed role: the agent command, in session mode. Comparing against the + *configured* implementer there describes a pairing that never happens -- + session mode never calls that route -- so the verdict was about two + things that never meet. + + Reported rather than refused: a single-model setup is a legitimate thing + to run deliberately, and blocking it would be the harness overruling an + operator about their own budget. What it must not be is a surprise. + """ + reviewer = roles.get("reviewer") + if implemented_by: + if reviewer is None: + return (True, "no implementer/reviewer pair configured") + return ( + True, + f"{reviewer.model} reviews work written by `{implemented_by}`, which this " + "harness does not route: the two are not the same model, and nothing here " + "knows which vendor is behind the agent", + ) + implementer = roles.get("implementer") + if reviewer is None or implementer is None: + return (True, "no implementer/reviewer pair configured") + if reviewer.model == implementer.model: + return ( + False, + f"reviewer and implementer are the same model ({reviewer.model}): " + "every review is a model grading its own work", + ) + if reviewer.provider.name == implementer.provider.name: + return ( + False, + f"reviewer and implementer share a provider ({reviewer.provider.name}): " + "reviews are independent of the model but not of the vendor", + ) + return (True, f"{reviewer.model} reviews {implementer.model}") + + +#: The cheapest question that proves a model is actually being served. An +#: endpoint advertising a model in `/models` is not the same claim. +PROBE_MESSAGES: tuple[Mapping[str, Any], ...] = ({"role": "user", "content": "ping"},) + + #: A transport is any callable that performs one request. Injected rather #: than imported so the retry logic is testable without a network, and so a #: caller can use whatever HTTP client it already has. @@ -257,37 +352,73 @@ def __init__( self.run_id = run_id or uuid.uuid4().hex[:12] self._seq = itertools.count() - def reviewer_independence(self) -> tuple[bool, str]: - """Whether the reviewer is actually independent of the implementer. + def reviewer_independence(self, implemented_by: str = "") -> tuple[bool, str]: + """Whether this client's reviewer is independent of the implementer. + + The logic is a free function so that a caller holding a *project's* + effective map -- which this client's own map may not be -- gets the + same answer from the same code. + """ + return reviewer_independence(self.roles, implemented_by=implemented_by) - Returns (independent, why). This was documented in three places and - enforced in none, which meant a reviewer could be the same model as - the implementer and nothing would say so -- some share of reviews - being a model grading its own work, invisibly. + def routed_by(self, routes_provider: Callable[[], Mapping[str, Route]]) -> ModelClient: + """A sibling client that resolves routes differently. - Reported rather than refused: a single-model setup is a legitimate - thing to run deliberately, and blocking it would be the harness - overruling an operator about their own budget. What it must not be is - a surprise. + Transport, retry policy, prices, telemetry and — deliberately — the + endpoint parks are shared: a spend cap belongs to the endpoint and + this process, not to whichever project happened to hit it first, and + a per-project copy of the parks would let every project rediscover + the same exhausted window at full price. + + The run id is *not* shared. Two clients emitting the same (run_id, + seq) pair would make two different attempts indistinguishable to a + consumer deduplicating a replayed stream, which is exactly what that + identity exists to prevent. """ + return ModelClient( + roles=routes_provider(), + transport=self.transport, + policy=self.policy, + on_event=self.on_event, + prices=self.prices, + sleep=self.sleep, + now=self.now, + jitter=self.jitter, + parks=self.parks, + routes_provider=routes_provider, + ) + + def answers(self, route: Route, *, timeout: float = 10.0) -> tuple[bool, str]: + """Whether a route's model replies at all, in one request. + + Deliberately not `call`: the ladder is six attempts with escalating + backoff, which is the ~20 minutes *per item* that discovering an + unusable model the expensive way costs. This asks once, briefly, and + reports what came back. + + The detail names the model and the status, because + "claude-sonnet-4-6 returned HTTP 504" names the thing to change and + "not ready" does not. No parking and no telemetry: a probe must not + idle an endpoint for the fleet, and a readiness poll is not a model + call anybody should find in their cost rollup. + """ + options = {**route.options, "max_tokens": 1, "timeout": timeout} try: - implementer = self.roles["implementer"] - reviewer = self.roles["reviewer"] - except KeyError: - return (True, "no implementer/reviewer pair configured") - if reviewer.model == implementer.model: + response = self.transport(route, PROBE_MESSAGES, options) + except Exception as exc: # noqa: BLE001 - any failure is the same answer return ( False, - f"reviewer and implementer are the same model ({reviewer.model}): " - "every review is a model grading its own work", + f"{route.model} could not be reached at {route.endpoint}: " + f"{type(exc).__name__}: {str(exc)[:160]}", ) - if reviewer.provider.name == implementer.provider.name: - return ( - False, - f"reviewer and implementer share a provider ({reviewer.provider.name}): " - "reviews are independent of the model but not of the vendor", - ) - return (True, f"{reviewer.model} reviews {implementer.model}") + if 200 <= response.status < 300: + return (True, f"{route.model} answered") + verdict = route.provider.classify(response.status, response.headers, response.body) + return ( + False, + f"{route.model} returned HTTP {response.status}" + + (f": {verdict.message[:160]}" if verdict.message else ""), + ) def route_for(self, role: str) -> Route: if self.routes_provider is not None: diff --git a/src/agent_harness/preflight.py b/src/agent_harness/preflight.py index fd49f96..0a7fca1 100644 --- a/src/agent_harness/preflight.py +++ b/src/agent_harness/preflight.py @@ -27,7 +27,7 @@ import tempfile import threading import time -from collections.abc import Callable +from collections.abc import Callable, Mapping from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -309,6 +309,133 @@ def probe() -> tuple[bool, str]: return probe +@dataclass(frozen=True) +class Answer: + """What one model said when asked a trivial question.""" + + ok: bool + detail: str + seconds: float + + +#: Asks one route's model for a minimal completion. Injected, like every +#: other probe here, so a readiness gate can be tested without a network. +Ask = Callable[[Any], Answer] + + +@dataclass +class RoleReachability: + """Which configured models actually answered, and which did not. + + Three buckets rather than a boolean because the middle one is a different + decision: a model that answers late is usable and must not be refused, + while a model that does not answer at all can complete nothing. + """ + + answered: dict[str, str] = field(default_factory=dict) + slow: dict[str, str] = field(default_factory=dict) + silent: dict[str, str] = field(default_factory=dict) + + def summary(self) -> str: + # Failures first: they are what the reader has to act on. + everything = {**self.silent, **self.slow, **self.answered} + return "; ".join(f"{role}: {detail}" for role, detail in everything.items()) + + +def remembered(ask: Ask, *, ttl: float = 60.0, now: Callable[[], float] = time.time) -> Ask: + """The same answer for a short while, per model and endpoint. + + Readiness is polled — a dashboard refreshing every few seconds would + otherwise turn a reachability probe into a load generator against the + model endpoint, and bill for it. Whether an endpoint serves a model does + not change between two polls seconds apart, so it is asked at most once a + minute per route. + """ + seen: dict[tuple[str, str], tuple[float, Answer]] = {} + + def cached(route: Any) -> Answer: + key = (str(getattr(route, "model", "")), str(getattr(route, "endpoint", ""))) + found = seen.get(key) + if found is not None and found[0] > now(): + return found[1] + answer = ask(route) + seen[key] = (now() + ttl, answer) + return answer + + return cached + + +def role_reachability_probe( + routes: Mapping[str, Any], + ask: Ask, + *, + timeout: float = 10.0, + patience: float = 20.0, +) -> Callable[[], RoleReachability]: + """Does each configured role's model actually answer? + + Preflight used to check that a reviewer was *routed*, which is true of a + model that will never reply. An endpoint can advertise a model in + `/models` and serve nothing behind it; the harness then discovered that + once per item, after the planner and implementer had been paid for, at + the cost of the whole retry ladder — six attempts with escalating backoff + is fifteen to twenty minutes of wall clock, per item, spent establishing a + condition that was true before the run started. + + In parallel, because the roles are independent and the point is to be + quick. `timeout` is the budget a healthy model is expected to meet; + `patience` is the deadline after which it is treated as no answer at all. + Between the two it is reported as slow and **not** failed: a model that + replies late is usable, and refusing it would be this gate overreaching. + + Daemon threads, deliberately: an ask that never returns must not hold the + report open, and must not hold the *process* open either. A pooled worker + is joined at interpreter exit, which would make one wedged endpoint delay + every shutdown of the service. + """ + + def probe() -> RoleReachability: + report = RoleReachability() + answers: dict[str, Answer] = {} + + def record(role: str, route: Any) -> None: + try: + answers[role] = ask(route) + except Exception as exc: # noqa: BLE001 - a probe reports, it does not raise + answers[role] = Answer(False, f"could not be asked: {str(exc)[:160]}", 0.0) + + threads = [ + threading.Thread( + target=record, args=(role, route), name=f"harness-probe-{role}", daemon=True + ) + for role, route in routes.items() + ] + for thread in threads: + thread.start() + deadline = time.monotonic() + patience + for thread in threads: + thread.join(max(0.0, deadline - time.monotonic())) + + for role, route in routes.items(): + model = str(getattr(route, "model", "?")) + answer = answers.get(role) + if answer is None: + report.silent[role] = f"{model} did not answer within {patience:g}s" + elif not answer.ok: + # Always named, whether or not the asker already did: a + # failure that does not say which model failed does not say + # what to change. + detail = answer.detail + report.silent[role] = detail if detail.startswith(model) else f"{model}: {detail}" + elif answer.seconds > timeout: + report.slow[role] = f"{answer.detail} after {answer.seconds:.0f}s" + else: + report.answered[role] = f"{answer.detail} in {answer.seconds:.1f}s" + return report + + return probe + + def disk_space_probe(path: str, floor_gb: float = 0.0) -> tuple[bool, str]: """Free space on the volume holding a project's checkout.""" try: @@ -331,6 +458,7 @@ def preflight_project( has_fleet: bool, reviewer_route: Any = None, reviewer_independent: tuple[bool, str] | None = None, + role_probe: Callable[[], RoleReachability] | None = None, session_host: Probe | None = None, git_probe: Callable[[str], tuple[bool, str]] = _is_git_repo, github_probe: Callable[[str], tuple[bool, str]] = _gh_can_write, @@ -410,6 +538,35 @@ def preflight_project( ) ) + if role_probe is not None: + reach = role_probe() + if reach.silent: + # Blocking, because a fleet whose reviewer cannot answer can + # complete nothing: every item runs to the last step and fails + # there, having already paid for the plan and the implementation. + checks.append( + Check( + "role reachability", + False, + f"{reach.summary()} — the model is configured but not answering, " + "so every item would fail at that stage after being paid for", + ) + ) + elif reach.answered or reach.slow: + checks.append(Check("role reachability", True, reach.summary())) + if reach.slow: + # A warning: a slow model is usable, and a preflight budget is not + # a statement about how long a real completion may take. + checks.append( + Check( + "model latency", + False, + "; ".join(f"{role}: {detail}" for role, detail in reach.slow.items()) + + " — slow to answer a one-token prompt, so every call pays that first", + blocking=False, + ) + ) + if reviewer_independent is not None: independent, why = reviewer_independent # A warning, not a blocker: running one model is a legitimate diff --git a/src/agent_harness/runtime.py b/src/agent_harness/runtime.py index 2b2c9ee..87e783b 100644 --- a/src/agent_harness/runtime.py +++ b/src/agent_harness/runtime.py @@ -20,11 +20,13 @@ from __future__ import annotations import shlex -from collections.abc import Callable +from collections.abc import Callable, Mapping +from dataclasses import dataclass from pathlib import Path from typing import Any from .executor import Checks +from .model_client import Route from .session_executor import AgentSpec, SessionExecutor from .work import Project, WorkQueue @@ -33,6 +35,53 @@ #: the fleet to describe its own return type. ExecutorFactory = Callable[[str], Any] +#: The only role a `SessionExecutor` routes to a model. Planning and +#: implementation are done by the agent process, with its own credentials and +#: its own endpoint, so a `planner` or `implementer` route is configuration +#: this executor will never call. +SESSION_EXECUTOR_ROLES = frozenset({"reviewer"}) + + +@dataclass(frozen=True) +class ExecutorRoles: + """Which configured roles the executor this deployment builds can reach. + + The role map is how an operator answers "what am I paying for, and what is + grading it?". In session mode it answered confidently and wrongly for two + of three stages: `planner` and `implementer` were advertised, editable and + echoed back, and nothing ever called them — so spend was looked for in an + audit log where it could never appear. + + `calls` of None means every configured role is called, which is the + non-session `Executor` and the honest default for a deployment that has + not said otherwise. + """ + + calls: frozenset[str] | None = None + #: What implements when implementation is not a routed role: the agent + #: argv, in session mode. Empty when the executor routes every stage. + implemented_by: str = "" + + @classmethod + def for_session(cls, agent: AgentSpec | None = None) -> ExecutorRoles: + command = tuple((agent or AgentSpec()).command) + return cls(calls=SESSION_EXECUTOR_ROLES, implemented_by=shlex.join(command)) + + def calls_role(self, role: str) -> bool: + return self.calls is None or role in self.calls + + def unused_reason(self, role: str) -> str: + """Why this deployment will never call `role`, in words. Empty when it will.""" + if self.calls_role(role): + return "" + if self.implemented_by: + return ( + f"the agent process (`{self.implemented_by}`) does the {role}'s work, " + "with its own credentials and endpoint; this route is only called by " + "the non-session executor" + ) + return "the active executor never calls this role" + class NotExecutable(RuntimeError): """This project cannot be executed as configured. @@ -61,6 +110,7 @@ def session_executor_factory( host: Any, agent: AgentSpec | None = None, reviewer: Any | None = None, + routes_for: Callable[[str], Mapping[str, Route]] | None = None, github_for: Callable[[str], Any] | None = None, ui_base_url: str = "", on_event: Callable[[dict[str, Any]], None] | None = None, @@ -72,6 +122,14 @@ def session_executor_factory( exercised end to end without a session host, a model or a network — which is the only way the wiring gets tested at all, since every real component here costs money or credentials to touch. + + `routes_for` resolves one project's **effective** role map — the global + map with that project's persisted overrides applied. Without it every + project shared the one global reviewer: a project could pass preflight on + its own reviewer override and then have its work reviewed by the global + model, or fail with `no route for role reviewer` when the global map had + none. It is a callable, consulted per call, so `PUT /api/roles` and a + change to the project row both still take effect without a restart. """ def build(project_id: str) -> SessionExecutor: @@ -89,7 +147,7 @@ def build(project_id: str) -> SessionExecutor: Path(project.work_dir), agent=agent or AgentSpec(), checks=_checks_for(project), - reviewer=reviewer, + reviewer=_reviewer_for(project_id), github=(github_for(project.repo) if github_for and project.repo else None), base_branch=project.base_branch, ui_base_url=ui_base_url, @@ -100,6 +158,17 @@ def build(project_id: str) -> SessionExecutor: executor.reap_orphaned_worktrees() return executor + def _reviewer_for(project_id: str) -> Any: + """This project's reviewer, not the deployment's. + + The client is rebuilt per executor rather than per call because the + route it resolves is already read live; what changes here is only + *whose* map it reads. + """ + if reviewer is None or routes_for is None: + return reviewer + return reviewer.routed_by(lambda: routes_for(project_id)) + return build diff --git a/src/agent_harness/schemas.py b/src/agent_harness/schemas.py index b15fa5c..2bc2fc7 100644 --- a/src/agent_harness/schemas.py +++ b/src/agent_harness/schemas.py @@ -180,19 +180,55 @@ class RoleRoute(BaseModel): ) +class RoutedRole(RoleRoute): + """A route, and whether this deployment's executor ever calls it.""" + + used: bool = Field( + True, + description="False when the active executor never calls this role, so the " + "route is configuration nothing acts on. In session mode the agent process " + "plans and implements with its own credentials and endpoint, and only the " + "reviewer is a routed model call -- an operator reading `implementer` here " + "would otherwise look for that spend in an audit log where it can never " + "appear.", + ) + unused_reason: str = Field( + "", description="What does this role's work instead, in words. Empty when `used`." + ) + + class RoleMap(BaseModel): + """The role map as it is set. Sent to `PUT /api/roles`.""" + + roles: dict[str, RoleRoute] = Field( + description="role -> where its calls go. Changing this takes effect on the next " + "call: the call site names a ROLE, never a model, which is what makes the map " + "changeable without a redeploy." + ) + + +class RoleMapView(BaseModel): + """The role map as it will actually be used. + + Returned by both the read and the write, because a `PUT` that stores a + route nothing calls should say so in the same breath rather than echo it + back as if it had changed what runs. + """ + reviewer_independent: bool = Field( True, description="False when the reviewer is the same model, or the same vendor, as " "the implementer -- some share of reviews is then a model grading its own work. " - "Reported rather than refused: running one model is a legitimate deliberate " - "choice, but it must not be a surprise.", + "Computed against the implementer the active executor *actually* uses, and by " + "the same code as preflight's `reviewer independence` check, so the two cannot " + "disagree. Reported rather than refused: running one model is a legitimate " + "deliberate choice, but it must not be a surprise.", ) reviewer_note: str = Field("", description="Why, in words.") - roles: dict[str, RoleRoute] = Field( - description="role -> where its calls go. Changing this takes effect on the next " - "call: the call site names a ROLE, never a model, which is what makes the map " - "changeable without a redeploy." + roles: dict[str, RoutedRole] = Field( + description="role -> where its calls go, and whether anything calls it. This is " + "the global map; a project may override any role, in which case its own " + "preflight reports the route that project will use." ) diff --git a/tests/test_deployment_docs.py b/tests/test_deployment_docs.py index 477634c..5dc97fc 100644 --- a/tests/test_deployment_docs.py +++ b/tests/test_deployment_docs.py @@ -19,7 +19,7 @@ import pytest from agent_harness.__main__ import main -from agent_harness.preflight import preflight_project +from agent_harness.preflight import RoleReachability, preflight_project from agent_harness.schemas import ExecutionReadiness, ProjectReadiness, ReadinessProbe from agent_harness.work import Project @@ -74,6 +74,7 @@ def test_every_blocker_the_document_tells_you_to_fix_can_actually_occur() -> Non has_fleet=False, reviewer_route=None, reviewer_independent=(False, "same vendor"), + role_probe=lambda: RoleReachability(silent={"reviewer": "m returned HTTP 504"}), session_host=lambda: (False, "refused"), checks_probe=lambda: (False, "base check failed"), disk_probe=lambda path, floor: (False, "disk is full"), @@ -91,9 +92,10 @@ def test_the_warnings_named_as_non_blocking_really_are() -> None: has_fleet=True, reviewer_route={"model": "m"}, reviewer_independent=(False, "same vendor"), + role_probe=lambda: RoleReachability(slow={"reviewer": "m answered after 14s"}), git_probe=lambda path: (True, path), github_probe=lambda repo: (True, repo), disk_probe=lambda path, floor: (True, "100 GiB free"), ) assert report.ready, "a start would be refused for reasons the document calls warnings" - assert {"checks", "reviewer independence"} <= {c.name for c in report.warnings} + assert {"checks", "reviewer independence", "model latency"} <= {c.name for c in report.warnings} diff --git a/tests/test_model_client.py b/tests/test_model_client.py index 9b9bd73..0409e98 100644 --- a/tests/test_model_client.py +++ b/tests/test_model_client.py @@ -19,6 +19,8 @@ RetryExhausted, RetryPolicy, Route, + effective_routes, + routes_from_map, ) MESSAGES = [{"role": "user", "content": "x"}] @@ -513,3 +515,105 @@ def test_clearing_an_endpoint_clears_every_role_on_it() -> None: assert parks.remaining("https://api", 1000.0, "implementer") == 0.0 assert parks.remaining("https://api", 1000.0, "reviewer") == 0.0 + + +# ----------------------------------------------- effective routes per project + + +def test_a_project_overrides_one_role_and_inherits_the_rest() -> None: + """The defect this prevents: the map was chosen wholesale, so a project + naming only a reviewer lost the global planner, and the executor ignored + the project map entirely.""" + merged = effective_routes( + {"planner": Route("global-planner", "https://g"), "reviewer": Route("global", "https://g")}, + {"reviewer": Route("project", "https://p")}, + ) + + assert merged["reviewer"].model == "project" + assert merged["planner"].model == "global-planner" + + +def test_a_route_missing_half_of_itself_is_not_a_route() -> None: + """Dropped, so preflight reports the role as unrouted rather than the + first call failing after the item is claimed and paid for.""" + routes = routes_from_map( + { + "reviewer": {"model": "m", "endpoint": "https://e", "provider": "generic"}, + "planner": {"model": "m", "endpoint": ""}, + } + ) + + assert set(routes) == {"reviewer"} + assert routes["reviewer"].provider is P.GENERIC + + +def test_a_sibling_client_routes_elsewhere_but_shares_the_parks() -> None: + """One process, one endpoint: a spend cap belongs to the endpoint, not to + whichever project hit it first, so a per-project client must not get a + fresh set of parks to rediscover it with.""" + transport = Recorder(ok(), ok()) + client = build(transport, roles={"reviewer": Route("global", "https://a")}) + + sibling = client.routed_by(lambda: {"reviewer": Route("project", "https://a")}) + + assert sibling.parks is client.parks + assert sibling.run_id != client.run_id + sibling.call("reviewer", MESSAGES) + client.call("reviewer", MESSAGES) + assert [c.model for c in transport.calls] == ["project", "global"] + + +# ---------------------------------------------------- does the model answer? + + +def test_a_model_that_answers_is_asked_exactly_once() -> None: + """Not `call`: the ladder is six attempts with escalating backoff, which + is the cost this probe exists to avoid paying once per item.""" + transport = Recorder(ok()) + client = build(transport, roles={"reviewer": Route("m", "https://a")}) + + answered, detail = client.answers(Route("m", "https://a"), timeout=5.0) + + assert answered is True + assert "m answered" in detail + assert len(transport.calls) == 1 + + +def test_an_advertised_but_unserved_model_is_reported_with_its_status() -> None: + """`claude-sonnet-4-6 returned HTTP 504` names the thing to change. + `not ready` does not.""" + client = build(Recorder(fail(504)), roles={"reviewer": Route("claude-sonnet-4-6", "https://a")}) + + answered, detail = client.answers(Route("claude-sonnet-4-6", "https://a")) + + assert answered is False + assert "claude-sonnet-4-6" in detail and "504" in detail + + +def test_a_transport_that_raises_is_an_answer_too() -> None: + """A probe reports; it never raises into the readiness gate that asked.""" + + def boom(route: Route, messages: Any, options: Any) -> Response: + raise TimeoutError("timed out") + + client = ModelClient(roles={}, transport=boom) + + answered, detail = client.answers(Route("m", "https://a")) + + assert answered is False + assert "TimeoutError" in detail + + +def test_the_probe_asks_for_one_token_and_names_its_own_deadline() -> None: + """A probe inheriting the work timeout would take ten minutes to + establish that a model is not answering.""" + seen: list[Mapping[str, Any]] = [] + + def record(route: Route, messages: Any, options: Any) -> Response: + seen.append(options) + return ok() + + ModelClient(roles={}, transport=record).answers(Route("m", "https://a"), timeout=7.0) + + assert seen[0]["max_tokens"] == 1 + assert seen[0]["timeout"] == 7.0 diff --git a/tests/test_preflight.py b/tests/test_preflight.py index d7baa83..7a2c3e3 100644 --- a/tests/test_preflight.py +++ b/tests/test_preflight.py @@ -8,12 +8,19 @@ from __future__ import annotations +import threading from pathlib import Path from typing import Any import pytest -from agent_harness.preflight import clean_checks_probe, preflight_project +from agent_harness.preflight import ( + Answer, + clean_checks_probe, + preflight_project, + remembered, + role_reachability_probe, +) from agent_harness.work import Project, WorkQueue @@ -143,6 +150,122 @@ def test_the_opt_in_clean_base_probe_blocks_on_the_first_failed_check(tmp_path: assert "`false` failed on base branch" in base.detail +# -------------------------------------------------- does the model answer? + + +class Model: + """A route, as much of one as a probe needs to name it.""" + + def __init__(self, model: str) -> None: + self.model = model + self.endpoint = "https://e" + + +def answering(seconds: float = 0.2) -> Any: + return lambda route: Answer(True, f"{route.model} answered", seconds) + + +def test_a_configured_model_that_never_answers_blocks_the_start() -> None: + """The reported cost: a reviewer that is advertised and not served passed + every gate, and each item then ran to completion and failed at the last + step -- six attempts with escalating backoff, per item, to establish a + condition that was true before the run started.""" + probe = role_reachability_probe( + {"reviewer": Model("claude-sonnet-4-6")}, + lambda route: Answer(False, f"{route.model} returned HTTP 504", 1.0), + ) + + report = run(project(), role_probe=probe) + + assert not report.ready + assert "claude-sonnet-4-6" in report.summary() and "504" in report.summary() + + +def test_a_model_that_is_merely_slow_warns_and_is_not_refused() -> None: + """A generous deadline and a warning. A late answer is still an answer, + and refusing one would be this gate overruling an operator about an + endpoint the harness does not own.""" + probe = role_reachability_probe( + {"reviewer": Model("slow-model")}, answering(seconds=14.0), timeout=10.0, patience=20.0 + ) + + report = run(project(), role_probe=probe) + + assert report.ready + assert [c.name for c in report.warnings] == ["model latency"] + assert "14s" in next(c for c in report.warnings if c.name == "model latency").detail + + +def test_a_model_that_answers_promptly_is_reported_and_passes() -> None: + report = run( + project(), role_probe=role_reachability_probe({"reviewer": Model("m")}, answering()) + ) + + assert report.ready + check = next(c for c in report.checks if c.name == "role reachability") + assert check.ok and "m answered" in check.detail + + +def test_a_request_that_never_returns_is_a_failure_not_a_hang() -> None: + """The report must not wait for the thing it is reporting on. A probe + thread that never returns would otherwise hold the HTTP request open for + as long as the model takes to not answer.""" + released = threading.Event() + + def never(route: Any) -> Answer: + released.wait(30.0) + return Answer(True, "eventually", 30.0) + + probe = role_reachability_probe({"reviewer": Model("wedged")}, never, patience=0.2) + try: + report = run(project(), role_probe=probe) + finally: + released.set() + + assert not report.ready + assert "wedged did not answer within 0.2s" in report.summary() + + +def test_the_roles_are_asked_in_parallel_not_one_after_another() -> None: + """Three roles, one deadline. Serial probes would multiply the worst case + by the number of roles, which is how a five-second refusal becomes a + request nobody waits for.""" + arrived = threading.Barrier(3, timeout=10.0) + + def wait_for_the_others(route: Any) -> Answer: + arrived.wait() + return Answer(True, f"{route.model} answered", 0.1) + + probe = role_reachability_probe( + {"planner": Model("a"), "implementer": Model("b"), "reviewer": Model("c")}, + wait_for_the_others, + ) + + assert run(project(), role_probe=probe).ready + + +def test_the_same_model_is_not_asked_again_on_the_next_poll() -> None: + """Readiness is polled. A probe on every poll would make a dashboard a + load generator against the model endpoint, and bill for it.""" + asked: list[str] = [] + clock = [1000.0] + + def ask_once(route: Any) -> Answer: + asked.append(route.model) + return Answer(True, f"{route.model} answered", 0.1) + + ask = remembered(ask_once, ttl=60.0, now=lambda: clock[0]) + probe = role_reachability_probe({"reviewer": Model("m")}, ask) + + probe() + probe() + assert asked == ["m"] + + clock[0] += 61.0 + probe() + assert asked == ["m", "m"] + + def test_project_registration_rejects_shell_check_syntax(client: Any) -> None: response = client.post( "/api/projects", diff --git a/tests/test_readiness.py b/tests/test_readiness.py index bd249df..637f6cf 100644 --- a/tests/test_readiness.py +++ b/tests/test_readiness.py @@ -98,14 +98,17 @@ def build( fleet: Any = None, host: Any = None, reviewer: dict[str, str] | None = REVIEWER, + roles: dict[str, dict[str, str]] | None = None, projects: tuple[Project, ...] = (), probes: dict[str, Any] | None = None, + model_client: Any = None, + executor_roles: Any = None, ) -> Iterator[Any]: queue = WorkQueue(str(tmp_path / "w.sqlite")) for p in projects or (project(),): queue.add_project(p) - if reviewer: - queue.set_setting(ROLE_MAP_KEY, {"reviewer": reviewer}) + if roles or reviewer: + queue.set_setting(ROLE_MAP_KEY, roles or {"reviewer": reviewer}) store = EventStore(tmp_path / "e.sqlite") with TestClient( create_api( @@ -115,6 +118,8 @@ def build( fleet=fleet, session_host=host, probes=OFFLINE if probes is None else probes, + model_client=model_client, + executor_roles=executor_roles, ) ) as client: holder: Any = client @@ -259,6 +264,177 @@ def test_readiness_needs_a_token(monitoring: Any) -> None: assert monitoring.get("/api/readiness").status_code == 401 +# ------------------------------------------------- whose reviewer is it anyway + + +def test_a_project_that_routes_its_own_reviewer_needs_no_global_one(tmp_path: Path) -> None: + """`ProjectSpec.roles` is persisted and documented as a per-project + override. Reading only the global map made the top-level answer contradict + the project report underneath it.""" + own = project(roles={"reviewer": {"model": "project-model", "endpoint": "https://p"}}) + client = next( + build(tmp_path, fleet=OneWorker(), host=ReadableHost(), reviewer=None, projects=(own,)) + ) + + body = client.get("/api/readiness", headers=hdr()).json() + + assert body["reviewer"]["ok"] is True + assert "every project routes its own" in body["reviewer"]["detail"] + assert body["projects"][0]["blockers"] == [] + + +def test_a_project_overriding_one_role_still_inherits_the_global_reviewer( + tmp_path: Path, +) -> None: + """A partial project map is an override of the roles it names, not a + replacement of the map. Treating it as a replacement refused a project + whose reviewer was routed all along.""" + partial = project(roles={"planner": {"model": "cheap", "endpoint": "https://a"}}) + client = next(build(tmp_path, fleet=OneWorker(), host=ReadableHost(), projects=(partial,))) + + body = client.get("/api/readiness", headers=hdr()).json() + + assert [c["name"] for c in body["projects"][0]["blockers"]] == [] + + +# --------------------------------------------- what the deployment will call + + +def session_mode() -> Any: + from agent_harness.runtime import ExecutorRoles + from agent_harness.session_executor import AgentSpec + + return ExecutorRoles.for_session(AgentSpec(command=("claude", "-p", "{prompt_file}"))) + + +THREE_ROLES = { + "planner": {"model": "gpt-5.6", "endpoint": "https://c", "provider": "claw-bay"}, + "implementer": {"model": "gpt-5.6", "endpoint": "https://c", "provider": "claw-bay"}, + "reviewer": {"model": "claude-sonnet-4-6", "endpoint": "https://c", "provider": "claw-bay"}, +} + + +def test_the_role_map_says_which_routes_the_executor_never_calls(tmp_path: Path) -> None: + """The map is how an operator answers "what am I paying for, and what is + grading it?". In session mode the agent process plans and implements with + its own credentials, so two of the three answers described a model that is + never asked anything -- and spend that would be looked for in an audit log + where it can never appear.""" + client = next(build(tmp_path, roles=THREE_ROLES, executor_roles=session_mode())) + + roles = client.get("/api/roles", headers=hdr()).json()["roles"] + + assert roles["reviewer"]["used"] is True + assert roles["implementer"]["used"] is False + assert "claude -p" in roles["implementer"]["unused_reason"] + # And the configuration is still there: the non-session executor uses it. + assert roles["implementer"]["model"] == "gpt-5.6" + + +def test_storing_a_route_nothing_calls_says_so_in_the_same_breath(tmp_path: Path) -> None: + """`PUT` used to succeed, echo the value back, and change nothing about + what runs.""" + client = next(build(tmp_path, executor_roles=session_mode())) + + body = client.put("/api/roles", headers=hdr(), json={"roles": THREE_ROLES}).json() + + assert body["roles"]["planner"]["used"] is False + assert body["roles"]["reviewer"]["used"] is True + + +def test_the_role_map_and_preflight_cannot_disagree_about_independence( + tmp_path: Path, +) -> None: + """Two endpoints, one property, two answers: `/api/roles` reported + `reviewer_independent: true` while preflight reported the same property as + `ok: false` -- from the *configured* implementer, which session mode never + calls.""" + client = next( + build( + tmp_path, + fleet=OneWorker(), + host=ReadableHost(), + roles=THREE_ROLES, + executor_roles=session_mode(), + ) + ) + + advertised = client.get("/api/roles", headers=hdr()).json() + checks = client.get("/api/projects/p/preflight", headers=hdr()).json()["checks"] + independence = next(c for c in checks if c["name"] == "reviewer independence") + + assert advertised["reviewer_independent"] == independence["ok"] + assert advertised["reviewer_note"] == independence["detail"] + # ...and it is about the thing that actually writes the code. + assert "claude -p" in independence["detail"] + + +# ------------------------------------------- does the configured model answer + + +class Unserved: + """A model client whose endpoint advertises a model it does not serve.""" + + def __init__(self, status: int = 504) -> None: + self.status = status + self.asked: list[str] = [] + + def answers(self, route: Any, *, timeout: float = 10.0) -> tuple[bool, str]: + self.asked.append(route.model) + return (False, f"{route.model} returned HTTP {self.status}") + + +class Serving(Unserved): + def answers(self, route: Any, *, timeout: float = 10.0) -> tuple[bool, str]: + self.asked.append(route.model) + return (True, f"{route.model} answered") + + +def test_a_reviewer_that_does_not_answer_refuses_the_start(tmp_path: Path) -> None: + """Preflight checked that a reviewer was *routed*, which is true of a + model that will never reply. Every item then ran to completion and failed + at the last step, having paid for the plan and the implementation.""" + client = next(build(tmp_path, fleet=OneWorker(), host=ReadableHost(), model_client=Unserved())) + + refused = client.post("/api/projects/p/start", headers=hdr()) + + assert refused.status_code == 409 + assert "claude-sonnet-4-6" in refused.json()["detail"] + assert "504" in refused.json()["detail"] + assert client.queue.control(project_id="p")[0] != "running" + + +def test_a_model_that_answers_leaves_the_start_alone(tmp_path: Path) -> None: + serving = Serving() + client = next(build(tmp_path, fleet=OneWorker(), host=ReadableHost(), model_client=serving)) + + body = client.get("/api/readiness", headers=hdr()).json() + + assert body["projects"][0]["ready_to_start"] is True + assert serving.asked == ["claude-sonnet-4-6"] + + +def test_only_the_roles_this_executor_calls_are_probed(tmp_path: Path) -> None: + """Spending tokens to ask whether a model the agent process replaces is + answering, and refusing a start over the answer, would be a gate about + nothing.""" + serving = Serving() + client = next( + build( + tmp_path, + fleet=OneWorker(), + host=ReadableHost(), + roles=THREE_ROLES, + model_client=serving, + executor_roles=session_mode(), + ) + ) + + client.get("/api/readiness", headers=hdr()) + + assert serving.asked == ["claude-sonnet-4-6"] + + # ------------------------------------------------------------- the probe diff --git a/tests/test_serve_fleet.py b/tests/test_serve_fleet.py index ddd5a6d..e0d4a66 100644 --- a/tests/test_serve_fleet.py +++ b/tests/test_serve_fleet.py @@ -29,7 +29,13 @@ from agent_harness.api import ROLE_MAP_KEY, create_api from agent_harness.audit import AuditStore from agent_harness.fleet import Fleet -from agent_harness.model_client import ModelClient, Response, Route +from agent_harness.model_client import ( + ModelClient, + Response, + Route, + effective_routes, + routes_from_map, +) from agent_harness.runtime import NotExecutable, session_executor_factory from agent_harness.session_executor import AgentSpec, SessionExecutor from agent_harness.session_host import IDLE, RUNNING, Session @@ -157,6 +163,99 @@ def test_building_creates_no_workers_and_claims_nothing(repo: Path, tmp_path: Pa assert queue.control(project_id="p")[0] == STOPPED +def test_each_project_is_reviewed_by_its_own_model(repo: Path, tmp_path: Path) -> None: + """`ProjectSpec.roles` is persisted and documented as a per-project + override, and preflight already decided on it. The factory injected the + one global client anyway, so a project could pass preflight on its own + reviewer and have the work reviewed by a different model -- or fail with + `no route for role reviewer` when the global map had none.""" + queue = WorkQueue(str(tmp_path / "w.sqlite")) + for project_id, model in (("a", "reviewer-a"), ("b", "reviewer-b")): + queue.add_project( + Project( + project_id=project_id, + name=project_id, + work_dir=str(repo), + roles={"reviewer": {"model": model, "endpoint": "https://own"}}, + ) + ) + + def routes_for(project_id: str) -> dict[str, Route]: + project = queue.get_project(project_id) + return effective_routes( + {"reviewer": Route("global-reviewer", "https://global", P.GENERIC)}, + routes_from_map((project.roles if project else None) or {}), + ) + + build = session_executor_factory( + queue, host=FakeHost(), reviewer=reviewer(), routes_for=routes_for + ) + + assert build("a").reviewer.route_for("reviewer").model == "reviewer-a" + assert build("b").reviewer.route_for("reviewer").model == "reviewer-b" + + +def test_a_project_without_an_override_still_gets_the_global_reviewer( + repo: Path, tmp_path: Path +) -> None: + queue = WorkQueue(str(tmp_path / "w.sqlite")) + queue.add_project(project_for(repo)) + + build = session_executor_factory( + queue, + host=FakeHost(), + reviewer=reviewer(), + routes_for=lambda _id: {"reviewer": Route("global-reviewer", "https://global", P.GENERIC)}, + ) + + assert build("p").reviewer.route_for("reviewer").model == "global-reviewer" + + +def test_serve_wires_the_project_map_all_the_way_to_the_executor( + repo: Path, tmp_path: Path +) -> None: + """The whole point of the override is that it reaches a model call. This + builds the fleet exactly as `serve` does, and asks the executor it would + hand a worker which model it will call.""" + queue = WorkQueue(str(tmp_path / "w.sqlite")) + queue.add_project( + Project( + project_id="p", + name="P", + work_dir=str(repo), + roles={"reviewer": {"model": "project-model", "endpoint": "https://project"}}, + ) + ) + queue.set_setting( + ROLE_MAP_KEY, + { + "reviewer": { + "model": "global-model", + "endpoint": "https://global", + "provider": "generic", + } + }, + ) + args = argparse.Namespace( + session_host="https://sessions.example", + reviewer="", + endpoint="", + events=tmp_path / "events.jsonl", + db=str(tmp_path / "w.sqlite"), + agent="claude -p {prompt_file}", + no_push=True, + poll=1.0, + ) + + fleet, _client, _host, roles = _fleet_for_serve(args, queue) + + assert fleet is not None and roles is not None + executor = fleet.executor_factory("p") + assert executor.reviewer.route_for("reviewer").model == "project-model" + # ...and the deployment knows the agent does the implementing. + assert roles.calls_role("reviewer") and not roles.calls_role("implementer") + + def test_a_project_with_no_checkout_is_refused_at_build_time(tmp_path: Path) -> None: """Rather than returning an executor that fails every item, which costs money to discover.""" @@ -311,7 +410,7 @@ def test_serve_event_sink_writes_live_telemetry_to_the_audit_store(tmp_path: Pat poll=1.0, ) - fleet, client, _host = _fleet_for_serve(args, queue, audit=audit) + fleet, client, _host, _roles = _fleet_for_serve(args, queue, audit=audit) assert fleet is not None and client is not None fleet.on_event(