diff --git a/docs/research/session-fork-plan.md b/docs/research/session-fork-plan.md new file mode 100644 index 00000000..d421b180 --- /dev/null +++ b/docs/research/session-fork-plan.md @@ -0,0 +1,450 @@ +# Claude and Codex Session Fork Plan + +**Status:** proposed + +**Scope:** add context-preserving fork support for `claude` and `codex` only. Other +backends remain unsupported until they expose an equivalent native operation. + +## Goal + +Allow a flow to branch an already-open conversation into an independent research +conversation: + +```python +# Called between completed parent turns. +research = parent.fork() +findings = await research.aturn("Collect read-only evidence for this question.") +``` + +The fork must preserve the parent conversation through one completed history +boundary, leave the parent usable and unchanged, and let the child run +independently. By default the child inherits the parent's effective backend, +model, provider, permissions, skills, tools, effort, machine and workspace. Any +override must be explicit and must state whether it invalidates cache equivalence. + +Forking must use the backend's native history operation. Replaying the transcript +into a new session is not an acceptable fallback: it changes the prompt shape and +does not preserve the prefix-cache condition. + +Two rules make the boundary unambiguous: + +1. `Session.fork()` performs the native fork eagerly, without sending a user + prompt. The returned child already has its backend id. +2. A fork is allowed only while the parent is idle and its selected boundary is + completed. A call while a parent turn is running raises a clear error. A flow + that needs a child during a later parent turn must prepare it before starting + that turn; a tool callback must never create its own fork while the parent is + waiting for that callback. + +## Phase 0: Native Prerequisites + +No flowverse implementation is started until these backend facts are verified in +the exact CLI versions supported by the release. + +### Claude probe + +- Verify that `--resume --fork-session --session-id ` creates and + persists the child without a user prompt. If Claude requires a prompt to create + the branch, Claude cannot satisfy the eager boundary contract in this release + and its fork capability remains disabled. +- Verify the child id, transcript isolation, effective model/effort/permission, + MCP configuration and behavior when the parent continues immediately after the + fork. +- Record the minimum Claude version that supports this operation. Older versions + must report an unavailable capability rather than fail after a flow has started. + +### Codex probe + +- Verify `thread/fork` and `lastTurnId` against the app-server schema, including a + fork from the latest completed turn and from an earlier completed turn. +- Verify that a second app-server process, using the same Codex home, provider + environment, machine and workspace, can load the parent thread by id and fork + it. It must not create a second container or remote mirror for the same child. +- Verify that the child server can run while the parent server is in a turn. A + single server remains serialized; the fork runtime must own a separate server + when overlap is requested. +- Record the minimum Codex version. A server without `thread/fork` must produce a + capability error, not a transcript replay. + +### Failure and retry probe + +Test a native operation whose response is lost after the backend creates a child. +The driver must not blindly retry and create an unknown number of branches. It +must either reconcile the child id from the backend or surface an explicit +unreconciled-fork error and record the orphan in telemetry. + +## Phase 1: Core Fork Contract + +This phase adds the flow-facing primitive and the internal fork context. It must +land before any flowverse helper. + +### Public contract + +```python +class Session(Protocol): + forks: ClassVar[bool] + + @property + def last_turn_id(self) -> str | None: ... + + def fork( + self, + *, + last_turn_id: str | None = None, + permission: str | None = None, + ) -> Session: ... +``` + +The semantics are fixed: + +- `fork()` takes no prompt and performs the native operation before returning. +- The parent must be opened, idle and not closed. An unopened parent, an active + parent turn, a non-completed `last_turn_id`, or a session already moved to a + different backend raises a clear error before a child is exposed. +- With `last_turn_id=None`, the child is forked through the parent's latest + completed turn. Codex also accepts an earlier completed turn. Claude forks the + complete conversation and raises `NotImplementedError` for a non-None boundary + instead of silently ignoring it. +- `last_turn_id` is inclusive. It is exposed as the latest completed backend turn + id where the backend provides one; Claude returns `None` because its native + operation has no intermediate boundary argument. A flow does not need private + driver state to name an intermediate Codex boundary. +- The child has a new backend id, lock, meter, steering state and lifecycle. Its + `last_turn_id` starts empty and its parent metadata is immutable. +- The fork context snapshots the effective model, provider/account, permission, + service tier, effort, environment, machine, workspace, selected skills and + offered tools at the boundary. Later parent reconfiguration, fallback or skill + changes do not change the child. +- `permission` is the only v1 fork override. `None` inherits the parent's + effective permission; an explicit value is passed through the native operation, + recorded in telemetry and sets `cache_equivalent=false` when it differs from the + parent. It must be one of the common permission values and must be rejected + before the native call when the backend cannot express it. Model, provider, + effort and workspace overrides are deferred. +- Pending parent steering, waiting prompts, interactive answers and flow-owned + prompting state are not copied. Hooks, watchers and the cycle are shared as + routing facilities and receive the child session; they are not copied as a + second mutable lifecycle. +- Error precedence is stable: an unopened/active/closed parent or invalid + boundary is a `RuntimeError`, an unsupported capability is a + `NotImplementedError`, and a native backend failure keeps the existing + `CalledProcessError` diagnostic. These errors are never hidden by `suppress`. +- A forked child never uses the normal cross-account or cross-backend fallback + path. Native fork errors and child turn errors are surfaced explicitly. The + parent retains its existing fallback behavior. +- Backends without native support expose `forks = False` and raise + `NotImplementedError` without creating a child. The capability may still be + disabled at runtime when the installed CLI fails the Phase 0 probe. + +### Flow checker and capability model + +- Add a flow-facing `Forks` marker, analogous to `Goal`, so a flow that forks + declares `Annotated[Agent, Forks]`. The runner rejects a selected backend + without `Session.forks` before the first turn. +- Add `forks` to the capability catalogue. The catalogue is descriptive; the + annotation and runtime check are what enforce the requirement. +- Add `fork` to the checker's session-returning call set (`new`, `clone`, `fork`) + so turns on a forked child remain tracked. A normal Python flow can use it; + atlas bodies still have no `Session` and cannot express a fork node. +- `Session.fork` is present in the structural Protocol for every backend, so the + API is not literally "visible only where forks is true". The contract is that + only capable backends pass the declared check and successfully execute it. +- Keep `hmz.flows.fork(name, into)` documented as the flow-directory copy + operation; it is unrelated to `Session.fork()`. + +### Lifecycle and telemetry contract + +Add a fork-specific journal event rather than overloading `cycle.opened`: + +```json +{ + "event": "forked", + "agent": "actor", + "backend": "codex", + "provider": "local", + "parent_session_id": "parent-thread", + "session_id": "child-thread", + "parent_key": "actor-codex@local-parent-thread", + "session_key": "actor-codex@local-child-thread", + "last_turn_id": "turn-7" +} +``` + +The cycle must register and link the child as soon as the native response gives +its id, even if the first child turn later fails. The `*_session_id` fields are +backend ids for diagnostics; the `*_key` fields are the cycle's fully qualified +relation keys. Usage is recorded on child turn completion in numeric fields only; +no prompt, transcript or file content is written. Existing readers ignore the new +event, while tracing maps the child session's `parent` to `parent_key`. + +The implementation must update the journal protocol, cycle reader/linker, +tracing collector/renderers and tests together. A single unspecified "telemetry +field" is not sufficient because fork creation, child failure and per-turn cache +usage happen at different times. + +## Phase 2: Backend Drivers + +### Claude implementation + +The ordinary command paths remain: + +```text +new: --session-id +resume: --resume +``` + +The one-time native fork path is: + +```text +--resume --fork-session --session-id +``` + +The driver must perform this operation eagerly, adopt the child id before any +child prompt, and then use ordinary `--resume ` commands. The fork +branch must never use `--session-id` with the parent or resume the parent without +`--fork-session`. + +The command builder uses the immutable fork context, not the agent's later mutable +config, for model, effort, permission mode, service tier, provider environment, +skills and MCP configuration. A child process has its own process and session +lock, so parent and child may overlap. Closing the child kills only its process. + +Required changes include `ClaudeCodeSession._command()`, an eager native-fork +operation, fork-context storage, child adoption before the first turn and the +forked journal event. Tests must cover a parent that continues immediately after +the fork and a child whose first turn fails. + +### Codex implementation + +Codex uses app-server JSON-RPC rather than spawning `codex fork`: + +```json +{ + "method": "thread/fork", + "params": { + "threadId": "", + "lastTurnId": "" + } +} +``` + +The child is created eagerly on a dedicated app-server process that shares the +parent's backend home, provider environment, machine, workspace and reference- +counted MCP bridge. It is not made through the public `AgentBase.clone()` path, +which intentionally drops hooks, tools and cycle state and may create a second +machine. The fork runtime forwards child events to the parent agent's watchers +and journal while keeping child locks, server stop and meter state independent. + +The response's `thread.id` is adopted immediately. The first child prompt then +uses the existing `turn/start` path with the child thread id. The child server is +never the parent's server for an overlapping child turn. If the configured +machine cannot provide two app-server processes over the same backend home, the +fork fails explicitly rather than creating a different context. + +`thread/fork` parameters omit inherited model/provider/service tier and permission +values. An explicit v1 `permission` override is sent using Codex's sandbox and +approval fields. The first child `turn/start` must use the effective values +returned by the fork response, rather than silently re-reading a parent agent +that may have been reconfigured. + +The app-server layer needs a request/notification router that can wait for a +thread-specific asynchronous operation without dropping unrelated notifications. +This is required for fork recovery and any later compaction capability; changes +are not limited to `CodexSession._thread()`. + +## Phase 3: Flow-Side Research Helper + +This phase adds a convenient delegation workflow on top of the core primitive. It +does not implement native fork logic in `humanfia/flowverse`. + +### Prepared child model + +A tool callback runs while the parent turn is active and therefore cannot call +`parent.fork()`. The flowverse helper must prepare a child between parent turns: + +```python +slot = ResearchFork.prepare( + parent, permission="read-only", timeout=60, max_output_chars=32000 +) +parent.offers([slot.tool]) +try: + answer = await parent.aturn("Use the research tool if evidence is needed.") +finally: + slot.close() # also unregisters the helper from the parent +``` + +`prepare()` performs the eager native fork and captures the completed boundary. +The callback only sends the child-specific research prompt to the already-created +child, waits for it, validates the result and returns a digest through the +existing MCP bridge. Before exposing the slot, the helper replaces the forked +child's inherited tool list with the explicit read-only allowlist (empty by +default). It is one-shot by default; a flow explicitly prepares a new slot for +another parent turn. The child does not offer the research tool itself, so +recursive delegation is bounded. + +The helper accepts only Claude and Codex sessions with `forks=True`. It refuses +to prepare a child from an unopened, active, moved or unsupported parent. The +parent's transcript, id, pending turn and future prompts are never used as the +child's mutable state. + +### Read-only policy + +`Session.fork()` v1 inherits the parent's effective permission by default. The +research helper explicitly requests `permission="read-only"`, so it can be used +with a bypass parent without weakening the parent's own session. This explicit +override invalidates the default cache-equivalence guarantee and is recorded as +such. + +Flow-owned callbacks execute in the flow process, outside the backend sandbox. +The helper therefore offers no parent tools by default. If a flow supplies tools, +it must pass an explicit read-only allowlist; names or prompt instructions are not +treated as proof that a callback has no side effects. The parent and child may +share the MCP socket, but only those allowlisted callbacks are advertised to the +child. + +### Deadline, result and cleanup rules + +- The helper has a monotonic deadline covering child turn execution and result + validation, plus a small cleanup grace period. `asyncio.wait_for()` alone is + insufficient because the current awaited-turn worker continues after task + cancellation. +- Claude cancellation closes and joins the child process. Codex cancellation + stops only the child fork runtime and its dedicated app server; it must never + call `parent_agent.stop()`. +- A timeout, native error, child failure, malformed result or cleanup failure is + returned as an MCP tool error with a stable error code and diagnostic. It is + never converted to an empty finding. +- The child result has a typed flowverse schema with a maximum serialized size. + Oversized or schema-invalid output is an error, not an unbounded digest passed + into the parent context. + +The helper continues to use the normal MCP tool-call loop. `Verdict(adds=...)` is +a hook channel and is not a substitute for a tool result. + +### Flowverse coordination + +The core humanize2 change and the flowverse change are released together. The +flowverse pins or declares the minimum humanize2 version that provides +`Session.fork`, `Forks` and the prepared-child behavior. Its tests use the public +flow interfaces and fake native drivers; they do not import `hmz.agents` or call +private driver methods. + +## Cache-Preserving Requirements + +The default fork must preserve the provider-visible common prefix: + +- fork only at a completed boundary captured eagerly; +- use the same backend, effective model, provider/account, service tier, + permissions, system/developer instructions, tool definitions and workspace; +- send only the child-specific prompt after the native fork; +- never replay the parent transcript in a newly constructed prompt; +- never use cross-backend fallback for a forked conversation. + +The provider still decides whether a request hits its cache. Tests assert the +invariants and retain reported usage; they do not claim that every provider must +return a hit. + +Claude already reports cache-read/cache-write usage. Codex's token usage contains +separate cached and non-cached input fields. Before implementation, define one +non-double-counting mapping into the common usage model, for example: + +```text +input = net-new input tokens +cache_read = cached input tokens +cache_write = cache-write input tokens +output = output tokens +``` + +The chosen mapping must be documented, tested against `Usage.total`, and written +to child telemetry. The phrase "map them if needed" is not sufficient for the +cache acceptance condition. + +## Fork Then Compact (Deferred Follow-up) + +Compaction is intentionally outside the current fork v1. Keep it as a separate +follow-up after fork lifecycle and usage accounting are stable: + +```python +child.compact() +``` + +The child must be forked and idle; compaction never touches the parent. Add a +`compacts: ClassVar[bool]` capability and reject unsupported backends explicitly. + +- **Claude:** send `/compact` as a child-only operation over the stream-json + protocol. An empty successful result and `system/compact_boundary` are success; + `--autocompact` is not a force operation. +- **Codex:** call `thread/compact/start` on the child server and wait for a + thread-specific completion. Accept both the legacy `thread/compacted` + notification and the current `contextCompaction` item delivered through + `item/completed`; the app-server router must not discard either form. + +Compaction is one summarization pass with its own reported usage. Later child +turns use the smaller summary prefix. A parent may be compacted before preparing +many children only through an explicit parent operation, since that changes the +parent's own context. + +## Non-Supported Backends + +`agy`, `cursor`, `dsh`, `grok`, `kimi`, `pi`, `qwen`, `opencode`, `mimo` and +`zcode` do not receive fork support in this plan. The ACP bridge and +`HumanSession` also expose `forks = False`. They retain their current session, +batch and native-subagent behavior. A request to fork one of them fails before a +child is created; there is no transcript-replay fallback. + +## Specification and Documentation Gate + +This plan does not edit any `SPEC.md`. Under the repository instructions, a +separate explicit request is required before `src/hmz/agents/SPEC.md` or +`src/hmz/flows/SPEC.md` is changed. Until that approval exists, the implementation +must be labelled experimental and the public reference docs must not describe +forking as part of the normative interface. Once approved, update the reference +interfaces, checker docs, tracing docs and flowverse documentation in the same +release. + +## Acceptance Matrix + +| Area | Acceptance condition | +| --- | --- | +| Native prerequisites | Supported CLI versions prove eager native fork, shared backend state and response-loss behavior. | +| API | `fork()` is eager, prompt-free, idle-only, returns a child id and honors the completed boundary contract. | +| Capability | `Annotated[Agent, Forks]` rejects a backend without native fork before its first turn. | +| Claude | Native `--resume + --fork-session + --session-id` creates an isolated child and later child turns resume that id. | +| Codex | Native `thread/fork` creates the child, honors inclusive `lastTurnId`, and uses a dedicated overlapping server. | +| Snapshot | Parent effective configuration and selected session state are frozen at the boundary; explicit permission overrides set `cache_equivalent=false`. | +| Isolation | Parent id, transcript, pending turn and future prompts are unchanged; parent may continue immediately. | +| Prepared helper | The MCP callback consumes a pre-created child and never forks while the parent turn is waiting. | +| Tools | Only explicitly allowlisted read-only flow callbacks are offered to a research child. | +| Concurrency | Claude child processes overlap; Codex child and parent overlap only on separate servers and otherwise serialize per server. | +| Timeout | Deadline cancellation stops and joins only the child runtime; no worker, process or server is left behind. | +| Errors | Unopened/active parent, invalid boundary, unsupported native operation, timeout, malformed result and unreconciled fork are explicit. | +| Fallback | A forked child never moves to another account or backend implicitly. | +| Cache | Prefix invariants hold and Claude/Codex cache fields use a tested, non-double-counting mapping. | +| Telemetry | Fork creation, parent relation, boundary, child failure and per-turn usage are linkable in cycle/tracing output. | +| Compact (follow-up) | Deferred from fork v1; when implemented, child-only compaction must work with current and legacy Codex completion signals while the parent remains unchanged. | +| Compatibility | Unsupported backends and old CLI versions reject fork without creating a child. | +| Integration | humanize2 tests, docs and the pinned `humanfia/flowverse` tests agree on the experimental or approved contract. | + +## Required Tests + +- fake Claude CLI: eager no-prompt fork, exact flags, parent continuation, child + resume and child failure; +- fake Codex app server: `thread/fork`, inclusive boundary, separate-server + overlap, notification routing and response-loss recovery; +- fork rejection for unopened, active, closed, moved and unsupported sessions; +- parent config/state mutation after fork does not change child behavior; +- fallback is disabled for forked children while parent fallback remains intact; +- prepared MCP helper returns a typed digest and never forks from inside the + active parent callback; +- timeout, cancellation, process/server reaping, malformed output, size limit, + recursion limit and deterministic MCP error responses; +- read-only parent/tool allowlist enforcement, including a callback that would + mutate state if it were incorrectly advertised; +- cycle/tracing relation and cache usage records for successful and failed child + turns; +- Codex cache field mapping and `Usage.total` accounting; +- follow-up only: optional Claude and Codex compaction, including + `contextCompaction` and legacy `thread/compacted` signals; +- flow checker capability declaration and all non-Claude/Codex compatibility + cases; +- the humanize2 suite and the flowverse suite against the released interface. diff --git a/docs/research/workflow-orchestration-literature-survey.md b/docs/research/workflow-orchestration-literature-survey.md new file mode 100644 index 00000000..6f886a2e --- /dev/null +++ b/docs/research/workflow-orchestration-literature-survey.md @@ -0,0 +1,112 @@ +# Literature Survey: Deterministic Program Orchestration vs. Agentic Tool-Calling Loops + +**Date:** 2026-08-31 +**Context:** Preliminary research for a paper proposing that deterministic Python programs orchestrating LLM calls consume fewer tokens than agents managing the same workflow via iterative tool calls. + +## Executive Summary + +**Core Finding:** The exact controlled experiment (deterministic orchestration vs. LLM-controlled loop, same model/tools/tasks, only control-flow owner varies) **has been published** in a narrow domain (COBOL→Python, arXiv:2605.09894, May 2026), showing up to 3.5× token reduction at comparable accuracy. However, multiple weaknesses in that work leave room for a stronger, multi-domain contribution. + +**Novelty Assessment:** A general framework demonstration across heterogeneous benchmarks, with proper statistical rigor, cache-aware cost accounting, and characterization of when the pattern wins vs. loses, remains unpublished. + +**Critical Threats:** +1. **Prompt caching counterargument**: Raw token counts may not translate to dollar savings if programmatic workflows break cache prefixes while agentic loops preserve them (verified: r=0.15 between token reduction and cost reduction in one study). +2. **Architecture explains only 0.5% of variance** in one multi-architecture evaluation, versus 27.8% for model choice (though this result lacks token measurements and has wide CIs). +3. **One empirical study shows ReAct beating a fixed workflow on both accuracy AND tokens** on open-ended QA (HotpotQA, 200 examples), though the workflow was minimal. + +--- + +## 1. Direct Prior Art (Closest Matches) + +### 1.1 Deterministic vs. LLM-Controlled Orchestration (COBOL Modernization) + +**Citation:** Naing Oo Lwin, Rajesh Kumar. "Deterministic vs. LLM-Controlled Orchestration for COBOL-to-Python Modernization." arXiv:2605.09894 [cs.SE], 11 May 2026. https://arxiv.org/abs/2605.09894 + +**System:** ATLAS (Autonomous Transpilation for Legacy Application Systems), single-agent. + +**Benchmark:** NIST COBOL85 Test Suite, 382 programs, avg ~1,200 LOC/program. + +**Experimental Design:** +- **Two arms within same framework:** + - Deterministic: Fixed stage-based pipeline; model does not pick tools, order, retries, or termination; predicates over system state decide branching; fallback strategies in predefined order. + - LLM-controlled: Model selects tools, execution order, repair strategies, termination decision; traces can diverge run-to-run. +- **Held constant:** Model architecture/version, temperature, seeds, system prompts, task instructions, formatting, source programs, test inputs, environment, timeouts, validation config, tool set (6 tools: `read_file`, `write_file`, `list_files`, `web_scrape`, `run_command`, `git`), interfaces, permissions. +- **Sole variable:** Execution control. + +**Models:** Claude-Sonnet-4-5, GPT-5.1-Codex-Max, Grok-Code-Fast-1. + +**Results (Table 1 — Computational Accuracy, Success Rate, P5-CA, CVaR₀.₁):** + +| Model | Orchestration | CA | SR | P5-CA | CVaR₀.₁ | +|---|---|---|---|---|---| +| Claude-Sonnet-4-5 | Deterministic | 0.966 | **0.902** | 0.959 | 0.953 | +| Claude-Sonnet-4-5 | LLM-controlled | 0.964 | **0.918** | 0.956 | 0.949 | +| GPT-5.1-Codex-Max | Deterministic | 0.969 | **0.910** | 0.962 | 0.956 | +| GPT-5.1-Codex-Max | LLM-controlled | 0.964 | **0.937** | 0.958 | 0.951 | +| Grok-Code-Fast-1 | Deterministic | 0.961 | **0.872** | 0.951 | 0.943 | +| Grok-Code-Fast-1 | LLM-controlled | 0.958 | **0.906** | 0.941 | 0.934 | + +- CA gaps: +0.002, +0.005, +0.003 (all <1pp). +- **Success Rate gaps: -0.016, -0.027, -0.034 — LLM-controlled wins on SR across all three models**, at larger margins than CA. + +**Token/Cost (prose only, not tabulated):** +- NC, SQ modules: LLM-controlled 1.75M–2.25M tokens; deterministic 400K–700K tokens. +- Headline claim: "up to 3.5×" reduction (note: endpoints imply 3.2–4.4×, internally inconsistent). +- SQ cost: LLM-controlled >$140/success vs. deterministic ~$40. + +**Limitations (§5.5, authors' own):** +- "Structured workflows with well-defined stages and validation **may amplify** deterministic advantages relative to exploratory tasks." +- "Where validators are weak, poorly specified, or absent, deterministic pipelines **may provide fewer advantages** than adaptive agentic approaches." +- COBOL→Python only; NIST suite omits production dependencies (JCL, CICS, VSAM, DBs, vendor dialects); correctness bounded by test-oracle coverage; conclusions depend on chosen models/prompts; cost excludes latency/infrastructure/engineering overhead. + +**Statistical Rigor Issues:** +- N (number of seeded runs) never stated. +- No confidence intervals, variance figures, or significance tests anywhere. +- Only two modules (NC, SQ) have token counts; no per-module table. + +**Relation to Our Claim:** +- **This is the closest published match**: same system, same controlled experiment, same conclusion. +- **Weaknesses leave room for stronger work**: (a) single narrow domain with explicit validators; (b) no statistical rigor; (c) SR result partially contradicts CA claim; (d) no cache-aware cost accounting; (e) no characterization of when the pattern wins vs. loses. + +--- + +### 1.2 Compiled AI (Zero-Runtime-LLM Workflows) + +**Citation:** Geert Trooskens et al. (9 authors, XY.AI Labs / Stanford / Cornell / Harvard). "Compiled AI: Deterministic Code Generation for LLM-Based Workflow Automation." arXiv:2604.05150 [cs.SE], 6 Apr 2026 (v2 31 Jul 2026). https://arxiv.org/abs/2604.05150 + +**Core Difference from Our Claim:** +- LLM invoked **once at compile time** to generate code; resulting workflow artifact runs with **zero LLM calls at execution time**. +- Our proposal: LLM remains in the runtime loop as a function call; determinism is about who controls flow, not about eliminating the model. + +**Benchmarks:** BFCL function-calling (n=400), DocILE document intelligence (n=5,680 invoices). + +**Results:** +- BFCL: 96% task completion, zero runtime tokens; break-even vs. runtime inference at ~17 transactions; **57× token reduction at 1,000 transactions**. +- DocILE: ties Direct LLM at 80.0% KILE, leads at 80.4% LIR (no accuracy loss). +- Security (n=135): 96.7% prompt-injection detection, 87.5% static code safety, zero false positives. + +**Citability:** Strong quantitative evidence but architecturally distinct (compile-once-run-forever vs. deterministic-loop-with-LLM-calls). + +--- + +### 1.3 LLM-as-Code (Framework Argument, No Token Experiment) + +**Citation:** Junjia Qi et al. "LLM-as-Code: Agentic Programming for Agent Harness." arXiv:2606.15874 [cs.AI, cs.SE], KDD 2026 AgenticSE Workshop. https://arxiv.org/abs/2606.15874 + +**Core Argument (fully aligned with our thesis):** +- Control flow (looping, branching, sequencing) belongs in deterministic program code. +- LLM demoted from orchestrator to callable subroutine; model retains freedom inside each call but cannot redirect execution path. +- Per-call context assembled from execution history DAG, so context size depends on call depth instead of growing with accumulated steps. + +**Empirical Evidence:** +- One computer-use agent case study. +- Reported outcome: "stability of long visual operation sequences" — **no token measurements**. +- Abstract explicitly states token argument is analytical/theoretical (O(depth) vs. O(steps) in prose, no formal complexity analysis). + +**Relation to Our Claim:** +- **Good news**: The thesis is published, but the evidence is missing — we can cite this as motivation and fill the gap it left. +- No controlled token experiment, no benchmark numbers, no head-to-head comparison. + +--- + +## 2. Classical Decoupling Baselines (Plan-Then-Execute) diff --git a/src/hmz/agents/__init__.py b/src/hmz/agents/__init__.py index 9ddd5d36..beae2ddc 100644 --- a/src/hmz/agents/__init__.py +++ b/src/hmz/agents/__init__.py @@ -12,6 +12,7 @@ WINDOW, AgentBase, CommandSessionBase, + ForkContext, Meter, SessionBase, StreamSessionBase, @@ -24,6 +25,7 @@ SERVICE_TIERS, AgentConfig, AgentDefaults, + Forks, Goal, Isolated, Remote, @@ -135,6 +137,8 @@ def driver(backend: str) -> tuple[type[AgentBase], type[AgentConfig]]: "DshSession", "Event", "Failed", + "ForkContext", + "Forks", "Goal", "GrokBuildAgent", "GrokBuildAgentConfig", diff --git a/src/hmz/agents/base.py b/src/hmz/agents/base.py index ff2ddb19..7b372239 100644 --- a/src/hmz/agents/base.py +++ b/src/hmz/agents/base.py @@ -21,7 +21,18 @@ from abc import ABC, abstractmethod from collections import Counter, deque from concurrent.futures import ThreadPoolExecutor -from typing import IO, TYPE_CHECKING, Any, ClassVar, Literal, Protocol, Self, overload +from dataclasses import dataclass +from typing import ( + IO, + TYPE_CHECKING, + Any, + ClassVar, + Literal, + Protocol, + Self, + cast, + overload, +) from .codenames import codename from .event import Event, Failed, Question, Stopped, Unrecoverable, Usage, say @@ -43,6 +54,9 @@ from .config import AgentConfig +_LIVE = object() + + class Journal(Protocol): """Where an agent writes down a session it opened, which is the run it is part of. @@ -55,6 +69,83 @@ def opened(self, agent: AgentBase, session: str) -> None: """Writes down a session one of the agents has just opened.""" ... + def forked( + self, + agent: AgentBase, + parent: str, + child: str, + last_turn_id: str | None = None, + *, + provider: str | None = None, + permission: str | None = None, + cache_equivalent: bool = True, + ) -> None: + """Writes down that one session branched into another, at a completed boundary.""" + ... + + def fork_lost( + self, + agent: AgentBase, + parent: str, + last_turn_id: str | None = None, + *, + provider: str | None = None, + ) -> None: + """Writes down a fork whose child id was lost with the response that made it. + + A native fork that fails after the backend may already have created the child is not + retried -- a retry would create another branch -- so the orphan is written down for a + person to reconcile rather than left to multiply in silence. + """ + ... + + def fork_usage(self, agent: AgentBase, session: str, usage: Usage) -> None: + """Writes numeric usage for one completed fork child turn.""" + ... + + def fork_failed(self, agent: AgentBase, session: str, error: str) -> None: + """Writes a bounded diagnostic for a fork child turn that failed.""" + ... + + +@dataclass(frozen=True, slots=True) +class ForkContext: + """The effective configuration a forked child was branched under, frozen at the boundary. + + A fork preserves the provider-visible prefix of the parent conversation, so the child's + first turn must run under the same backend, model, account, permission, service tier, + effort, skills and tools the parent had when it was forked -- not whatever the parent has + since been reconfigured to. These are that, snapshotted where the fork was made. A later + `permission` override is the one field a flow may move; it is recorded separately, on the + child, and marks the fork as not cache-equivalent. + + Attributes: + model: The model the parent's turns ran at. + effort: How hard those turns thought, in the backend's own wording. + permission: The rung the parent ran at, one of :data:`PERMISSIONS`. + service_tier: The common provider tier the parent asked for. + provider: The account those turns ran as, or "" for this machine's own. + provider_ref: The immutable provider snapshot used to start child processes. + goals: Whether the parent's backend goal feature was switched on. + web_search: Whether the parent was allowed to search the web. + skills: The flow's skills the parent carried, by name. + tools: The flow's callbacks the parent offered, by name. + """ + + model: str + effort: str + permission: str + service_tier: str + provider: str + provider_ref: Provider + anchor: AnchorConfig | None + goals: bool + web_search: bool + skills: tuple[str, ...] + tools: tuple[str, ...] + cache_equivalent: bool = True + permission_override: bool = False + def _tee( source: IO[str], @@ -379,6 +470,13 @@ class SessionBase(ABC): #: offering it -- a tool the model never sees is a flow that does not do what it says. takes_tools: ClassVar[bool] = False + #: Whether this backend has a native history operation that branches a conversation in + #: place, which is what :meth:`fork` reaches for. Only Claude and Codex do; every other + #: backend keeps its current session behaviour and refuses a fork before a child is made. + #: A fact of the backend, said the same way `shapes` and `takes_tools` are -- a stand-in + #: written for a test says it too, `forks: ClassVar[bool] = True`. + forks: ClassVar[bool] = False + def __init__( self, agent: AgentBase, cwd: str | os.PathLike[str] | None = None ) -> None: @@ -401,6 +499,10 @@ def __init__( #: far as writing the tests wants the skill about writing them and no longer wants #: the eight about reading the codebase, and it is the same conversation either way. self._skills: tuple[str, ...] | None = None + #: A forked child's loaded skill objects, frozen at the branch boundary. The agent's + #: flow may load a different set later, but this conversation must keep the exact + #: objects it branched with so that a changed skill file cannot alter its prefix. + self._fork_loaded: tuple[Loaded, ...] | None = None #: The flow's own callbacks this conversation is putting in front of the agent, which #: it may say again between any two turns. Held here as well as in the agent's #: toolbox so that a session can say what it is offering without asking the agent @@ -421,6 +523,27 @@ def __init__( #: What this conversation is to think at from its next turn on, where it has been #: told something other than what its agent runs at, and None where it has not. self._effort: str | None = None + #: The frozen configuration a forked child was branched under, or None for a session + #: nobody forked. A forked child reads its model, effort, permission and the rest off + #: this rather than off its agent, so that reconfiguring or moving the parent later + #: does not change the child. + self._fork_context: ForkContext | None = None + #: A forked child gets its own callback bridge. The agent toolbox is intentionally shared + #: by ordinary sessions, but using it here would expose a parent's later tools to the + #: child (including the research callback that is waiting on it). + self._fork_toolbox: Toolbox | None = None + #: The parent's full agent config at the boundary, frozen by the config dataclass's own + #: immutability: `reconfigure` replaces rather than mutates it, so holding this is a + #: snapshot. Backend-specific fields a forked child's command builder reads -- Claude's + #: `allowed_tools`, Codex's `overrides` -- come off this, not the agent's live config. + self._frozen_config: AgentConfig | None = None + #: The backend's id for the last completed turn of this conversation, or None before + #: one has finished and for a backend whose native fork has no intermediate boundary. + self._last_turn: str | None = None + #: Every completed backend turn in order. Codex can fork through an earlier boundary; + #: keeping the ids here lets the public method reject an in-flight or unknown id before + #: it reaches the native driver. + self._completed_turns: tuple[str, ...] = () #: What this conversation has cost and how fast, written as the backend says what #: each request came to rather than once the turn is over: a turn is minutes long, #: and a rate that stood still for all of them would be a rate of nothing. @@ -481,7 +604,9 @@ def skills(self) -> tuple[str, ...]: flow's to say, and a name nothing answers to is a name to correct rather than a skill to invent. """ - brought = self._agent.loaded + brought = ( + self._fork_loaded if self._fork_loaded is not None else self._agent.loaded + ) if self._skills is None: return tuple(one.name for one in brought) wanted = set(self._skills) @@ -541,7 +666,7 @@ def offers(self, tools: Iterable[Tool] | None) -> None: f"{self._agent.backend} has no way of being given a tool of a flow's own" ) self._tools = () if tools is None else tuple(tools) - box = self._agent.toolbox + box = self._fork_toolbox or self._agent.toolbox if self._tools and self._unoffering is None: # Registered the first time anything is said and not before: the toolbox is keyed # by a number this session answers to for as long as it is alive, so what takes @@ -552,6 +677,24 @@ def offers(self, tools: Iterable[Tool] | None) -> None: # its number can be handed to another one. self._unoffering = weakref.finalize(self, box.offers, id(self), ()) box.offers(id(self), self._tools) + self._tools_changed() + if not self._tools and self._fork_toolbox is not None: + # A native fork may have started its private bridge while inheriting the parent's + # tools. Close that socket as soon as a helper narrows the child to no callbacks; + # Toolbox can lazily reopen it if the child is explicitly offered tools later. + box.close() + + def _toolbox(self) -> Toolbox: + """The callback bridge this session's backend must be given. + + Ordinary sessions share their agent's bridge because some backends run one server for + every conversation. A forked session owns a private bridge, which is what keeps the + parent's later tools (especially a callback waiting on the child) out of its context. + """ + return self._fork_toolbox or self._agent.toolbox + + def _tools_changed(self) -> None: # noqa: B027 -- empty hook for backend runtimes + """Lets a backend restart a child runtime whose tool configuration is immutable.""" @property def id(self) -> str: @@ -577,6 +720,18 @@ def named(self) -> str | None: """ return self._id + @property + def last_turn_id(self) -> str | None: + """The backend's id for the latest completed turn, where it exposes one. + + What a fork names its boundary by: forking through the latest completed turn means + naming this, and a backend that also takes an earlier one lets a flow name that + instead. None for a backend whose native fork has no intermediate boundary -- Claude + forks the whole conversation -- and before any turn has completed. A forked child + starts empty here, so that a child forked again names the child's own turns. + """ + return self._last_turn + def spent(self) -> Usage: """What this conversation has cost so far, by the kind of token it went on. @@ -645,6 +800,8 @@ def effort(self) -> str: through an answer, and a flow that changed it mid-turn would be describing a turn that never happened. """ + if self._fork_context is not None: + return self._fork_context.effort return self._effort or self._agent.effort @effort.setter @@ -709,7 +866,7 @@ def __call__[T: BaseModel]( # takes is that long on the next round too. raise except subprocess.CalledProcessError: - if not suppress: + if self._fork_context is not None or not suppress: raise return None if schema is not None else "" if schema is None: @@ -717,7 +874,7 @@ def __call__[T: BaseModel]( try: return _shaped(said, schema) except ValueError: - if not suppress: + if self._fork_context is not None or not suppress: raise return None @@ -835,7 +992,11 @@ def _turning( # Anything said while nobody was working goes into this turn. A flow's own prompt is # the only way into a turn that has not started, so it is asked for here rather than # written to the session: a session between turns would answer it on its own. - held = self._agent.waiting() if self._agent.waiting is not None else [] + held = ( + self._agent.waiting() + if self._fork_context is None and self._agent.waiting is not None + else [] + ) if held: prompt = "\n\n".join([prompt, *held]) # Before the moments rather than only on the first turn: a session closed and then @@ -867,33 +1028,46 @@ def _turning( # for. On the prompt as it is sent rather than on the one the hooks and the # transcript see, which is the flow's own words: a schema in the transcript # is the plumbing showing through. - for event in self._falling_back(prompt, schema=schema): - if event.kind == "result": - # Held back: a hook may yet send the agent on, and a turn that was - # sent on has not answered. - answered = event - continue - self._heard(event) - if event.kind == "tool": - named, _, about = event.text.partition(" ") - self._fire(Moment.PRE_TOOL_USE, tool=named, about=about) - elif event.kind in ("subagent", "subagent-ends"): - # An agent this one started of its own, bracketed the way a turn is: - # a fleet under a turn is something a flow may want a word about, and - # the id is what makes the one that started and the one that ended - # one agent rather than two lines. - named, _, about = event.text.partition(" ") - self._fire( - Moment.SUBAGENT_START - if event.kind == "subagent" - else Moment.SUBAGENT_STOP, - tool=named, - about=about, - under=event.whose, - ) - yield event + try: + for event in self._falling_back(prompt, schema=schema): + if event.kind == "result": + # Held back: a hook may yet send the agent on, and a turn that was + # sent on has not answered. + answered = event + continue + self._heard(event) + if event.kind == "tool": + named, _, about = event.text.partition(" ") + self._fire(Moment.PRE_TOOL_USE, tool=named, about=about) + elif event.kind in ("subagent", "subagent-ends"): + # An agent this one started of its own, bracketed the way a turn is: + # a fleet under a turn is something a flow may want a word about, + # and the id is what makes the one that started and the one that + # ended one agent rather than two lines. + named, _, about = event.text.partition(" ") + self._fire( + Moment.SUBAGENT_START + if event.kind == "subagent" + else Moment.SUBAGENT_STOP, + tool=named, + about=about, + under=event.whose, + ) + yield event + except BaseException as failed: + if self._fork_context is not None and self._agent.cycle is not None: + with contextlib.suppress(Exception): + self._agent.cycle.fork_failed( + self._agent, self._id or "", type(failed).__name__ + ) + raise # Heard whether or not it is passed on, because what a turn cost is on it. self._heard(answered) + if self._fork_context is not None and self._agent.cycle is not None: + with contextlib.suppress(Exception): + self._agent.cycle.fork_usage( + self._agent, self._id or "", answered.spent + ) stopping = self._fire( Moment.STOP, said=answered.text, prompt=prompt, again=again ) @@ -942,6 +1116,13 @@ def _falling_back( that failed for a reason no other try could come out differently on is a turn that has failed, and trying it again is a loop rather than a recovery. """ + if self._fork_context is not None: + # A native fork is a single backend/account lineage. Moving it through the normal + # fallback chain would destroy the provider-visible prefix and make its child id + # meaningless, so failures are surfaced directly from the fork runtime. + yield from self._stream(self._shaped_ask(prompt, schema), schema=schema) + return + from hmz import fallbacks, providers # How this place is tried again, read once for the whole walk: it is a file, and a @@ -1282,6 +1463,214 @@ def unsteered(self, text: str) -> None: del self._steered[ticket] return + def fork( + self, *, last_turn_id: str | None = None, permission: str | None = None + ) -> SessionBase: + """Branches this conversation into an independent one, preserving its prefix. + + The fork is eager and prompt-free: the native operation runs before this returns, the + returned child already has its backend id, and the parent is left exactly as it was -- + open, idle, its transcript and pending turn and future prompts untouched -- so it may + be driven on at once while the child runs on its own. Only an open, idle, unmoved + session may be forked, and only Claude and Codex have a native operation for it. + + Args: + last_turn_id: The completed turn to fork through, inclusive. None forks through the + latest completed turn; Codex also accepts an earlier completed turn, and Claude + raises NotImplementedError for a non-None boundary. + permission: The rung the child runs at, or None to inherit the parent's. The one v1 + override; an explicit value is passed through the native operation and marks the + fork as not cache-equivalent where it differs from the parent. + + Returns: + The child session, already named by the backend. + + Raises: + NotImplementedError: If this backend has no native fork. + RuntimeError: If the parent is unopened, running a turn, closed, or moved to + another backend. + ValueError: If `permission` is not one of the common rungs. + """ + from .config import PERMISSIONS + + if not type(self).forks: + raise NotImplementedError(f"{type(self).__name__} has no native fork") + if self._working: + raise RuntimeError("cannot fork while a turn is running") + # Do not wait behind a turn: a callback running inside that turn could otherwise block + # here while the backend waits for the callback to return. A second check under the + # session lock closes the race with a turn starting immediately after the first check. + if not self._lock.acquire(blocking=False): + raise RuntimeError("cannot fork while a turn is running") + try: + if self._working: + raise RuntimeError("cannot fork while a turn is running") + if self._id is None: + raise RuntimeError("session has not run a turn yet") + if self._ended: + raise RuntimeError("cannot fork a closed session") + if self._moved_to is not None: + raise RuntimeError( + "cannot fork a session that moved to another backend" + ) + if permission is not None and permission not in PERMISSIONS: + raise ValueError( + f"permission must be one of {', '.join(PERMISSIONS)}, " + f"not {permission!r}" + ) + self._validate_fork_boundary(last_turn_id) + ready = getattr(type(self), "native_ready", None) + if callable(ready) and not ready(): + raise NotImplementedError( + f"{type(self).__name__} native fork is unavailable in this CLI" + ) + child = type(self)(self._agent, self._cwd) + child._fork_from(self, last_turn_id=last_turn_id, permission=permission) + return child + finally: + self._lock.release() + + def _validate_fork_boundary(self, last_turn_id: str | None) -> None: + """Rejects a boundary that is not one of this session's completed turns.""" + if last_turn_id is not None and last_turn_id not in self._completed_turns: + raise RuntimeError( + f"last_turn_id {last_turn_id!r} is not a completed turn of this session" + ) + + def _fork_from( + self, parent: SessionBase, *, last_turn_id: str | None, permission: str | None + ) -> None: + """Snapshots the parent, performs the native fork, and names this child. + + Args: + parent: The conversation being branched, which this child is made from. + last_turn_id: The boundary, as for :meth:`fork`. + permission: The override, as for :meth:`fork`. + """ + source = parent._fork_context + source_config = parent._frozen_config or parent._agent.config + effective = ( + permission + if permission is not None + else (source.permission if source is not None else source_config.permission) + ) + # `fork` refuses an unopened parent before this is reached, so the parent has an id. + assert parent._id is not None # noqa: S101 + from dataclasses import replace + + provider = source.provider_ref if source is not None else parent._agent.node() + provider = replace( + provider, + env=dict(provider.env), + args=tuple(provider.args), + ) + self._fork_toolbox = Toolbox() + weakref.finalize(self, self._fork_toolbox.close) + self._fork_loaded = tuple(parent._carrying()) + offered = tuple(parent._toolbox().offered()) + self._frozen_config = source_config + self._fork_context = ForkContext( + model=source.model if source is not None else source_config.model, + effort=source.effort if source is not None else parent.effort, + permission=effective, + service_tier=( + source.service_tier + if source is not None + else source_config.service_tier + ), + provider=provider.name, + provider_ref=provider, + anchor=source.anchor if source is not None else parent._agent.anchor, + goals=(source.goals if source is not None else parent._agent.goals_enabled), + web_search=( + source.web_search if source is not None else source_config.web_search + ), + skills=parent.skills, + tools=tuple(one.name for one in offered), + cache_equivalent=(source.cache_equivalent if source is not None else True) + and ( + permission is None + or effective + == ( + source.permission + if source is not None + else source_config.permission + ) + ), + permission_override=permission is not None, + ) + # The child carries the parent's skills and callbacks, frozen: its own answer, so that + # a later `loads` or `offers` on the parent does not move it. + self._skills = tuple(self._fork_context.skills) + self._tools = offered + if self._tools: + self.offers(self._tools) + try: + child_id = self._fork(parent_id=parent._id, last_turn_id=last_turn_id) + except BaseException: + # The child has not been exposed, so release its private bridge if native fork + # creation failed. A backend that already created an orphan records that separately. + self._shut() + self._fork_toolbox.close() + self._fork_toolbox = None + raise + self._adopt_fork(child_id, parent._id, last_turn_id) + + def _fork(self, *, parent_id: str, last_turn_id: str | None) -> str: + """The backend's native fork, already bound to a snapshot, returning the child id. + + The effective rung is already on :attr:`_fork_context`, so the backend reads it there + rather than being handed it again. + + Args: + parent_id: The parent's backend id, which the native operation branches from. + last_turn_id: The boundary, or None for the latest completed turn. + + Returns: + The child's backend id. + + Raises: + NotImplementedError: For a boundary this backend cannot express -- Claude has no + intermediate boundary -- and for a backend with no native fork. + subprocess.CalledProcessError: If the native operation fails. + """ + raise NotImplementedError(f"{type(self).__name__} has no native fork") + + def _adopt_fork( + self, child_id: str, parent_id: str, last_turn_id: str | None + ) -> None: + """Takes the child's backend id, and writes the fork down as a fork rather than an open. + + A forked child is a session the agent has opened, so it is in the agent's list; but the + run writes it as a branch of the parent rather than as a session that started from + nothing, so that a trace can draw the two as one lineage. + + Args: + child_id: The id the native fork gave this child. + parent_id: The parent's id, which the run links this child back to. + last_turn_id: The boundary, for the record. + """ + self._id = child_id + self._agent._opens(child_id) + if self._agent.cycle is not None: + self._agent.cycle.forked( + self._agent, + parent_id, + child_id, + last_turn_id, + provider=( + self._fork_context.provider + if self._fork_context is not None + else None + ), + permission=( + self._fork_context.permission if self._fork_context else None + ), + cache_equivalent=( + self._fork_context.cache_equivalent if self._fork_context else True + ), + ) + def close(self) -> None: """Ends the conversation, so that a turn under way stops waiting. @@ -1307,7 +1696,7 @@ def close(self) -> None: # another thread while this one may be offering: a list landing between the two # lines above would otherwise be an entry with nothing left to take it back. # Offering after this has returned is offering afresh, which registers its own. - self._agent.toolbox.offers(id(self), ()) + (self._fork_toolbox or self._agent.toolbox).offers(id(self), ()) if self._moved_to is not None: # And the conversation this one moved to, which is this conversation carried on # somewhere else: it ends when this one does. @@ -1319,6 +1708,9 @@ def close(self) -> None: # files would have them taken away underneath it. The turn itself lets go of them # as it ends, which is the first moment nothing is using them. self._unmounted() + if self._fork_toolbox is not None: + self._fork_toolbox.close() + self._fork_toolbox = None def _unmounted(self) -> None: """Takes away what this session mounted, once and whenever the last holder is done. @@ -1351,7 +1743,12 @@ def elsewhere(self) -> bool: Whether to let go of it before this turn. False for a session holding nothing yet, and for one whose agent has not moved. """ - return self._as is not None and self._as != self._agent.node().name + account = ( + self._fork_context.provider_ref.name + if self._fork_context is not None + else self._agent.node().name + ) + return self._as is not None and self._as != account @property def cwd(self) -> str: @@ -1360,7 +1757,11 @@ def cwd(self) -> str: Which is the path on the machine the work lands on: the one the session was opened with, or the workspace the flow is running in where it was opened with none. """ - anchor = self._agent.anchor + anchor = ( + self._fork_context.anchor + if self._fork_context is not None + else self._agent.anchor + ) if self._cwd is not None: return os.path.abspath(self._cwd) # noqa: PTH100 if anchor is not None: @@ -1383,7 +1784,11 @@ def _workspace(self) -> str: agent whose turns land elsewhere -- at one outside the workspace the anchor names. Said before the first turn rather than as a backend failing to start in it. """ - anchor = self._agent.anchor + anchor = ( + self._fork_context.anchor + if self._fork_context is not None + else self._agent.anchor + ) if anchor is None: where = self.cwd if not os.path.isdir(where): # noqa: PTH112 @@ -1456,7 +1861,9 @@ def _carrying(self) -> tuple[Loaded, ...]: Returns: The flow's own, in the flow's order, less any this session was told not to carry. """ - brought = self._agent.loaded + brought = ( + self._fork_loaded if self._fork_loaded is not None else self._agent.loaded + ) if self._skills is None: return tuple(brought) wanted = set(self._skills) @@ -1474,8 +1881,22 @@ def _environment(self) -> Mapping[str, str]: Returns: The variables to add, which are set for the turn and for nothing else. """ + if self._fork_context is not None: + return self._fork_context.provider_ref.env return self._agent.environment() + def _hushed(self) -> frozenset[str]: + """The credential variables to remove for this session's provider snapshot.""" + if self._fork_context is None: + return self._agent.hushed() + from hmz.backends import named + + profile = named(self._agent.backend) + if profile is None: + return frozenset() + provider = self._fork_context.provider_ref + return profile.accounts() - set(provider.env) + def _environ(self) -> dict[str, str] | None: """The whole environment this session's processes are started with. @@ -1483,13 +1904,25 @@ def _environ(self) -> dict[str, str] | None: This process's own, less what a provider hushes and plus what this session and that provider set, or None where there is nothing to change. """ - added, hushed = self._environment(), self._agent.hushed() + added, hushed = self._environment(), self._hushed() if not added and not hushed: return None return { name: value for name, value in os.environ.items() if name not in hushed } | dict(added) + def _spawned(self, argv: list[str]) -> list[str]: + """Spawns a command using this session's frozen backend context when forked.""" + if self._fork_context is None: + return self._agent.spawned(argv, self.cwd) + return self._agent.spawned( + argv, + self.cwd, + provider=self._fork_context.provider_ref, + anchor=self._fork_context.anchor, + toolbox=self._toolbox(), + ) + def _adopt(self, session_id: str) -> None: """Takes the name the backend gave this session, the first time a turn lands in it. @@ -1532,14 +1965,19 @@ def pursue(self, objective: str, *, suppress: bool = False) -> str: set: a flow that disabled them retains control of every continuation. subprocess.CalledProcessError: If the turn fails and `suppress` is not set. """ - if not self._agent.goals_enabled: + goals = ( + self._fork_context.goals + if self._fork_context is not None + else self._agent.goals_enabled + ) + if not goals: raise RuntimeError(f"{self._agent.id}: goals are disabled") try: return self._pursue(objective) except Unrecoverable: raise # not covered by `suppress`, for the reason it is not in a turn except subprocess.CalledProcessError: - if not suppress: + if self._fork_context is not None or not suppress: raise return "" @@ -1632,7 +2070,7 @@ def _stream( # signal handling with it, which a flow pumping turns from threads of its own has # no way to lend it. Whether there is one to spawn -- an anchor, a provider's own # paths, both -- is the agent's to say. - argv = self._agent.spawned(argv, self.cwd) + argv = self._spawned(argv) out: list[str] = [] err: list[str] = [] said: queue.Queue[Event | None] = queue.Queue() @@ -1653,7 +2091,16 @@ def _stream( env=self._environ(), # The directory the session was opened at, which is this one unless it was # opened at another; an anchored turn is put there by the anchor instead. - cwd=None if self._agent.anchor is not None else where, + cwd=( + None + if ( + self._fork_context.anchor + if self._fork_context is not None + else self._agent.anchor + ) + is not None + else where + ), ) as proc: assert proc.stdout is not None # noqa: S101 assert proc.stderr is not None # noqa: S101 @@ -1871,7 +2318,11 @@ def _stream( complained = "".join(self._complaints) self._shut() raise Failed(status or 1, argv, said, complained) - if self._agent.anchor is not None: + if ( + self._fork_context.anchor + if self._fork_context is not None + else self._agent.anchor + ) is not None: # An anchored turn has to be over when it says it is: coganchor pushes what the # agent wrote when the session ends, so a process held open past the turn would # leave that turn's work still on this machine. The cost is that an anchored @@ -1997,8 +2448,12 @@ def _start(self, argv: list[str]) -> subprocess.Popen[str]: # somewhere else is one that never starts again -- the wrong credentials for good. # Read early, the same fallback makes it start again once for nothing, which is a # turn's cost rather than a run's. - account = self._agent.node().name - argv = self._agent.spawned(argv, self.cwd) + account = ( + self._fork_context.provider_ref.name + if self._fork_context is not None + else self._agent.node().name + ) + argv = self._spawned(argv) started = subprocess.Popen( argv, stdin=subprocess.PIPE, @@ -2012,7 +2467,16 @@ def _start(self, argv: list[str]) -> subprocess.Popen[str]: env=self._environ(), # And in the directory the session was opened at, as for one of those: a backend # held open across its turns is held open where its conversation is rooted. - cwd=None if self._agent.anchor is not None else where, + cwd=( + None + if ( + self._fork_context.anchor + if self._fork_context is not None + else self._agent.anchor + ) + is not None + else where + ), ) assert started.stderr is not None # noqa: S101 # Which account it was started as, so that an agent that falls back is an agent whose @@ -2858,7 +3322,15 @@ def _environ(self) -> dict[str, str] | None: name: value for name, value in os.environ.items() if name not in hushed } | dict(added) - def spawned(self, argv: list[str], cwd: str = "") -> list[str]: + def spawned( + self, + argv: list[str], + cwd: str = "", + *, + provider: Provider | object | None = _LIVE, + anchor: AnchorConfig | object | None = _LIVE, + toolbox: Toolbox | object | None = _LIVE, + ) -> list[str]: """One turn of this agent, as the command to actually spawn. Every backend renders its own call and then comes here, so that what a turn is wrapped @@ -2875,6 +3347,9 @@ def spawned(self, argv: list[str], cwd: str = "") -> list[str]: or "" for the workspace itself. An anchored turn is told there rather than put there: the agent is started in this machine's mirror of that directory, which is the anchor's to work out. + provider: The provider snapshot to use, or the live agent provider when omitted. + anchor: The machine snapshot to use, or the live agent anchor when omitted. + toolbox: The callback bridge to expose, or the live agent toolbox when omitted. Returns: The command to spawn, which is `argv` itself for an agent that is neither anchored @@ -2890,22 +3365,33 @@ def spawned(self, argv: list[str], cwd: str = "") -> list[str]: # still fails saying what could not be found. if (found := elsewhere(argv[0])) is not None: argv = [found, *argv[1:]] - provider = self.provider - if provider is not None and provider.args: - argv = [*argv, *provider.args] - swaps = provider.swaps() if provider is not None else () + selected_provider = ( + self.provider if provider is _LIVE else cast("Provider | None", provider) + ) + if selected_provider is not None and selected_provider.args: + argv = [*argv, *selected_provider.args] + swaps = selected_provider.swaps() if selected_provider is not None else () # What the provider hands the agent as variables is the agent's own, and the target # is not to be given it: everything the agent exports is inherited by every command # it runs there, and a key crossing to another machine is a key on that machine. - private = tuple(provider.env) if provider is not None else () - anchor = self.anchor - if anchor is not None: - return self._reaching(anchor).command( + private = tuple(selected_provider.env) if selected_provider is not None else () + selected_anchor = ( + self.anchor if anchor is _LIVE else cast("AnchorConfig | None", anchor) + ) + selected_toolbox = ( + self._toolbox if toolbox is _LIVE else cast("Toolbox | None", toolbox) + ) + if selected_anchor is not None: + return self._reaching(selected_anchor, toolbox=selected_toolbox).command( argv, swaps=swaps, private=private, chdir=cwd ) - return provider.command(argv) if provider is not None else argv + return ( + selected_provider.command(argv) if selected_provider is not None else argv + ) - def _reaching(self, anchor: AnchorConfig) -> AnchorConfig: + def _reaching( + self, anchor: AnchorConfig, *, toolbox: Toolbox | object | None = _LIVE + ) -> AnchorConfig: """The anchor, plus whatever a turn under it has to be able to reach on this machine. Which is the bridge to the flow's own callbacks, for an agent that is offering any: @@ -2915,6 +3401,8 @@ def _reaching(self, anchor: AnchorConfig) -> AnchorConfig: Args: anchor: Where this agent's turns land. + toolbox: The callback bridge whose command must run on this machine, or None when no + callbacks are offered. Returns: It, or a copy naming the bridge as a program that runs here. The socket itself needs @@ -2923,9 +3411,12 @@ def _reaching(self, anchor: AnchorConfig) -> AnchorConfig: """ from dataclasses import replace - if self._toolbox.empty(): + selected_toolbox = ( + self._toolbox if toolbox is _LIVE else cast("Toolbox | None", toolbox) + ) + if selected_toolbox is None or selected_toolbox.empty(): return anchor - held = self._toolbox.command()[0] + held = selected_toolbox.command()[0] if held in anchor.local_execs: return anchor return replace(anchor, local_execs=(*anchor.local_execs, held)) diff --git a/src/hmz/agents/claude.py b/src/hmz/agents/claude.py index 5297f1a8..6c6077cb 100644 --- a/src/hmz/agents/claude.py +++ b/src/hmz/agents/claude.py @@ -2,7 +2,10 @@ from __future__ import annotations +import functools import json +import shutil +import subprocess import uuid from collections import Counter from dataclasses import dataclass @@ -10,7 +13,7 @@ from .base import AgentBase, StreamSessionBase from .config import AgentConfig -from .event import Event, Question, Usage +from .event import Event, Failed, Question, Usage from .hooks import EVERYWHERE, SUBAGENTS, Moment if TYPE_CHECKING: @@ -46,6 +49,7 @@ _ALLOWED_TOOLS_MAX = 32 _ALLOWED_TOOL_RULE_MAX_CHARS = 4096 +_FORK_SECONDS = 30.0 #: Reasons that leave an answer unfinished even when a broken intermediary labels the result #: `success`. Claude normally keeps its own agent loop going for these rather than returning @@ -89,6 +93,34 @@ } +@functools.lru_cache(maxsize=8) +def _native_fork_ready(binary: str | None = None) -> bool: + """Whether the installed Claude CLI advertises prompt-free session forking.""" + binary = binary or shutil.which("claude") + if binary is None: + return False + try: + result = subprocess.run( + [binary, "--help"], + stdin=subprocess.DEVNULL, + capture_output=True, + encoding="utf-8", + errors="replace", + timeout=5, + check=False, + ) + except (OSError, subprocess.TimeoutExpired): + return False + text = f"{result.stdout}\n{result.stderr}" + if "--fork-session" in text and "--session-id" in text: + return True + # Test doubles and third-party wrappers may not implement --help. Their native operation + # remains the source of truth, while a real Claude help page has a recognizable usage. + return result.returncode == 0 and not any( + marker in text for marker in ("Usage:", "Options:") + ) + + def _about(called: dict[str, Any]) -> str: """What a tool was called with, as the one line a row of a transcript has room for. @@ -175,6 +207,16 @@ class ClaudeCodeSession(StreamSessionBase): #: this turn without anything of the person at this machine's being written. takes_tools: ClassVar[bool] = True + #: `--resume --fork-session --session-id ` is Claude's native fork: it + #: branches a conversation in place without a prompt, so the child keeps the parent's + #: prefix for the provider's cache. A fork is eager and prompt-free here. + forks: ClassVar[bool] = True + + @classmethod + def native_ready(cls) -> bool: + """Whether this installation exposes Claude's native fork flags.""" + return _native_fork_ready(shutil.which("claude")) + def __init__( self, agent: AgentBase, cwd: str | os.PathLike[str] | None = None ) -> None: @@ -227,7 +269,7 @@ def _command(self) -> list[str]: # A fresh id per attempt: an opening turn that failed may still have left Claude holding # the id it was given, and retrying under that one would collide forever. pinned = self._id or str(uuid.uuid4()) - argv = [ + return [ "claude", "--print", "--input-format", @@ -237,20 +279,44 @@ def _command(self) -> list[str]: "--verbose", "--resume" if self._id else "--session-id", pinned, + *self._session_args(), + ] + + def _session_args(self) -> list[str]: + """The settings a process is started with, off the config in force for this session. + + A forked child reads them off its frozen fork context rather than off the agent, so + that reconfiguring the parent afterwards does not change the child -- and the fork + command itself is built from the same frozen values, so the branch carries them. + + Returns: + The permission, tier, model, effort, tool rules and MCP config, as argv. + """ + context = self._fork_context + if context is None: + config = self._agent.config + model, effort = self._agent.config.model, self.effort + permission = self._agent.config.permission + service_tier = self._agent.config.service_tier + goals = self._agent.goals_enabled + web_search = self._agent.config.web_search + else: + config = self._frozen_config + model, effort = context.model, context.effort + permission, service_tier = context.permission, context.service_tier + goals, web_search = context.goals, context.web_search + argv = [ *( ["--permission-mode", mode] - if (mode := _PERMITTED.get(self._agent.config.permission)) + if (mode := _PERMITTED.get(permission)) else ["--dangerously-skip-permissions"] ), "--settings", - json.dumps( - {"fastMode": self._agent.config.service_tier == "fast"}, - separators=(",", ":"), - ), + json.dumps({"fastMode": service_tier == "fast"}, separators=(",", ":")), "--model", - self._agent.config.model, + model, "--effort", - self.effort, + effort, ] if self._shaping is not None: # Claude validates the answer against this itself, so a turn that lands has @@ -263,13 +329,13 @@ def _command(self) -> list[str]: # scheduler -- and one told not to search the web is refused the two that reach it. # Everything else it may reach for is what its permission rung says it may. denied: list[str] = [] - if not self._agent.goals_enabled: + if not goals: denied += _CONTINUATION_TOOLS - if not self._agent.config.web_search: + if not web_search: denied += _WEB_TOOLS if denied: argv += ["--disallowedTools", ",".join(denied)] - allowed_tools = getattr(self._agent.config, "allowed_tools", ()) + allowed_tools = getattr(config, "allowed_tools", ()) if allowed_tools: argv += ["--allowedTools", ",".join(allowed_tools)] # Read once and kept, so that what the process is recorded as having been told is @@ -283,10 +349,117 @@ def _command(self) -> list[str]: # own servers away for the length of this flow, which is not this flow's to do. argv += [ "--mcp-config", - json.dumps(self._agent.toolbox.config(), separators=(",", ":")), + json.dumps(self._toolbox().config(), separators=(",", ":")), ] return argv + def _permission(self) -> str: + """The permission rung frozen for this child, or the live agent rung otherwise.""" + if self._fork_context is not None: + return self._fork_context.permission + return self._agent.config.permission + + def _fork_command(self, parent_id: str, child_id: str) -> list[str]: + """The one-time native fork: resume the parent and branch it into a named child. + + Args: + parent_id: The parent conversation, as Claude logged it. + child_id: The child the branch is to become, chosen up front. + + Returns: + The command, which must not be sent a user prompt: the fork is done by the flags. + """ + return [ + "claude", + "--print", + "--input-format", + "stream-json", + "--output-format", + "stream-json", + "--verbose", + "--resume", + parent_id, + "--fork-session", + "--session-id", + child_id, + *self._session_args(), + ] + + def _fork(self, *, parent_id: str, last_turn_id: str | None) -> str: + """Performs Claude's native fork eagerly, without sending a prompt. + + The branch is made by the flags alone; the child id is adopted before any turn, and + later child turns resume it with an ordinary ``--resume ``. + + Args: + parent_id: The parent conversation to branch. + last_turn_id: Refused: Claude forks the whole conversation, with no boundary. + + Returns: + The child's id, which the child adopts. + + Raises: + NotImplementedError: For a non-None boundary, which Claude cannot express. + subprocess.CalledProcessError: If the fork could not be made. + """ + if last_turn_id is not None: + raise NotImplementedError( + "Claude forks the whole conversation; it has no intermediate boundary" + ) + child_id = str(uuid.uuid4()) + argv = self._spawned(self._fork_command(parent_id, child_id)) + with subprocess.Popen( + argv, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + encoding="utf-8", + errors="replace", + bufsize=1, + env=self._environ(), + cwd=( + None + if ( + self._fork_context.anchor + if self._fork_context is not None + else self._agent.anchor + ) + is not None + else self._workspace() + ), + ) as proc: + assert proc.stdout is not None # noqa: S101 + assert proc.stderr is not None # noqa: S101 + assert proc.stdin is not None # noqa: S101 + try: + # An empty input closes stdin without a prompt. Calling communicate after a + # manual close would make Python flush the already-closed pipe a second time. + stdout, stderr = proc.communicate(input="", timeout=_FORK_SECONDS) + except subprocess.TimeoutExpired as timed: + proc.kill() + stdout, stderr = proc.communicate() + raise Failed( + 124, + argv, + stdout or "", + f"Claude native fork timed out after {_FORK_SECONDS:g}s: {stderr or ''}", + ) from timed + status = proc.returncode + if status != 0: + if self._agent.cycle is not None: + self._agent.cycle.fork_lost( + self._agent, + parent_id, + last_turn_id, + provider=( + self._fork_context.provider + if self._fork_context is not None + else None + ), + ) + raise Failed(status or 1, argv, stdout or "", stderr or "") + return child_id + def _write(self, text: str, ticket: str = "") -> str: """Renders one thing to say as the user message Claude reads it as. @@ -333,7 +506,14 @@ def _offered(self) -> tuple[str, ...]: argument schema again before every turn to catch a reworded sentence would cost each turn more than the sentence is worth. """ - return tuple(sorted(one.name for one in self._agent.toolbox.offered())) + return tuple(sorted(one.name for one in self._toolbox().offered())) + + def _validate_fork_boundary(self, last_turn_id: str | None) -> None: + """Claude's native fork has no intermediate turn boundary.""" + if last_turn_id is not None: + raise NotImplementedError( + "Claude forks the whole conversation; it has no intermediate boundary" + ) def _stale(self) -> bool: """Whether the process up was started for something this turn is no longer. @@ -563,7 +743,7 @@ def _answer(self, said: dict[str, Any]) -> None: about=_about(called), called=called, ) - if self._agent.config.permission == "read-only": + if self._permission() == "read-only": self._reply( said, {"behavior": "deny", "message": f"{tool} would change something"}, diff --git a/src/hmz/agents/codex.py b/src/hmz/agents/codex.py index 54c1acd2..4b6341c2 100644 --- a/src/hmz/agents/codex.py +++ b/src/hmz/agents/codex.py @@ -14,23 +14,28 @@ from __future__ import annotations import contextlib +import functools import itertools import json import os import queue +import shutil import signal import subprocess import sys +import tempfile import threading import weakref from collections import Counter -from dataclasses import dataclass +from dataclasses import dataclass, replace +from pathlib import Path from typing import TYPE_CHECKING, Any, ClassVar, cast from .base import AgentBase, SessionBase from .config import PERMISSIONS, AgentConfig from .event import Event, Failed, Question, Usage, say from .hooks import EVERYWHERE, SUBAGENTS, Moment, Occasion +from .tools import Toolbox if TYPE_CHECKING: from collections.abc import Callable, Iterator, Mapping, Sequence @@ -41,6 +46,38 @@ #: none of them is already a field of AgentConfig -- model, effort and permission are asked #: elsewhere, and a second place for them would be two answers. _OVERRIDE_KEYS = frozenset({"model_context_window", "model_auto_compact_token_limit"}) +_LIVE = object() + + +@functools.lru_cache(maxsize=8) +def _native_fork_ready(binary: str | None = None) -> bool: + """Whether the installed Codex app-server schema exposes ``thread/fork``.""" + binary = binary or shutil.which("codex") + if binary is None: + return False + try: + with tempfile.TemporaryDirectory(prefix="hmz-codex-probe-") as output: + result = subprocess.run( + [binary, "app-server", "generate-json-schema", "--out", output], + stdin=subprocess.DEVNULL, + capture_output=True, + encoding="utf-8", + errors="replace", + timeout=8, + check=False, + ) + files = tuple(path for path in Path(output).rglob("*") if path.is_file()) + schema = "\n".join( + path.read_text(encoding="utf-8", errors="replace") for path in files + ) + except (OSError, subprocess.TimeoutExpired): + return False + if "thread/fork" in schema: + return True + # Test doubles and wrappers often accept app-server but do not implement schema generation; + # let their native request decide while rejecting a real successful schema without the method. + return result.returncode == 0 and not schema and not result.stdout.strip() + #: What the server calls a turn stopping to ask its user something. Every other request it #: makes of a client is an approval, which an unattended flow does not stop for. @@ -58,6 +95,8 @@ class _Running: thread: str | None = None turn: str | None = None + #: The id of the latest completed turn, which is what a fork names its boundary by. + last: str | None = None #: The server this turn is being taken on, bound where the turn starts rather than asked #: for again when there is a word to put in. The agent's server is let go of and started #: again whenever what it was started knowing has moved -- another account, another list @@ -95,10 +134,53 @@ class _Running: "bypass": {"approvalPolicy": "never", "sandbox": "danger-full-access"}, } -#: What each kind of token is called in the totals the server states. Cached input is counted -#: inside the input rather than beside it, so it is not a kind of its own here: adding it would -#: be counting the same tokens twice. -_KINDS = {"input": "inputTokens", "output": "outputTokens"} + +def _usage(usage: Mapping[str, Any]) -> Counter[str]: + """Codex's token totals, as the common kinds, cached input counted once as `cache_read`. + + The server states the input with the cached part already inside it, so the two must not + both become kinds of their own -- that would count the cached tokens twice. `input` here + is the net-new input, `cache_read` the cached part, and `output` what came out: the three + together are the whole of what crossed the wire, so ``Usage.total`` still says that. + + Args: + usage: The ``tokenUsage.total`` the server stated. + + Returns: + The kinds, empty where the server said nothing. + """ + input_tokens = int(usage.get("inputTokens") or 0) + cached = int(usage.get("cachedInputTokens") or 0) + held = Counter( + { + "input": max(input_tokens - cached, 0), + "cache_read": cached, + "output": int(usage.get("outputTokens") or 0), + } + ) + return Counter({kind: tokens for kind, tokens in held.items() if tokens}) + + +def _permission_from_fork(result: Mapping[str, Any], default: str) -> str: + """Maps Codex's effective fork sandbox back into the common permission ladder.""" + sandbox: object = cast("object", result.get("sandbox")) + raw_type = cast( + "object", + cast("dict[str, object]", sandbox).get("type") + if isinstance(sandbox, dict) + else sandbox, + ) + sandbox_type = raw_type if isinstance(raw_type, str) else "" + if sandbox_type == "dangerFullAccess": + return "bypass" + if sandbox_type == "readOnly": + return "read-only" + if sandbox_type == "workspaceWrite": + policy = result.get("approvalPolicy") + if policy == "on-request": + return "auto" + return "workspace-write" + return default def unattended(permission: str, service_tier: str = "default") -> dict[str, Any]: @@ -258,6 +340,9 @@ def __init__(self, argv: list[str], env: Mapping[str, str] | None = None) -> Non #: total, so what one turn cost is the rise across it. self._counted: dict[str, Counter[str]] = {} self._messages: queue.Queue[dict[str, Any] | None] = queue.Queue() + #: Messages that arrived while another operation was waiting for its own response. + #: They are replayed to the next matching consumer instead of being silently discarded. + self._deferred: list[dict[str, Any]] = [] # Read from a thread of its own, so that a turn can wait on the server for a while # rather than only for as long as it takes. threading.Thread(target=self._pump, daemon=True).start() @@ -293,7 +378,9 @@ def permitted(self, permission: str, service_tier: str) -> dict[str, Any]: """ return unattended(self._instead.get(permission, permission), service_tier) - def call(self, method: str, params: dict[str, Any]) -> Any: + def call( + self, method: str, params: dict[str, Any], *, fallback: bool = True + ) -> Any: """Makes one call and reads until it is answered. A call naming a rung this machine's Codex will not take is made again a rung down: an @@ -304,6 +391,8 @@ def call(self, method: str, params: dict[str, Any]) -> Any: Args: method: The method to call. params: What to call it with. + fallback: Whether permission refusals may step down the normal agent ladder. Fork + creation passes False because a child may not change its effective sandbox. Returns: What the server answered with. @@ -312,6 +401,8 @@ def call(self, method: str, params: dict[str, Any]) -> Any: subprocess.CalledProcessError: If it refused the call for anything but the rung it named, or stopped before answering. """ + if not fallback: + return self._called(method, params) while True: try: return self._called(method, params) @@ -372,12 +463,11 @@ def _called(self, method: str, params: dict[str, Any]) -> Any: self._write( {"jsonrpc": "2.0", "id": ident, "method": method, "params": params} ) - # An answer is a message with no method of its own: the server asks things of us - # over the same stream, numbering its own calls, and one of those is not this one. - while (message := self._read()) is None or not ( - message.get("id") == ident and "method" not in message - ): - pass + # An answer is a message with no method of its own: notifications and replies for + # other operations share this stream, so the router keeps them for their consumer. + message = self._matching( + lambda one: one.get("id") == ident and "method" not in one + ) return self._answer(message, "") def pursue(self, params: dict[str, Any]) -> str: @@ -419,7 +509,13 @@ def pursue(self, params: dict[str, Any]) -> str: ) idle = False said = "" - while (message := self._read(_QUIET_SECONDS if idle else None)) is not None: + while ( + message := self._read_relevant( + thread=str(params["threadId"]), + ident=ident, + timeout=_QUIET_SECONDS if idle else None, + ) + ) is not None: if message.get("id") == ident and "method" not in message: self._answer(message, said) match message.get("method"): @@ -475,14 +571,19 @@ def turn(self, params: dict[str, Any], running: _Running) -> Iterator[Event]: costing = Usage() started: set[Any] = set() # the items this turn has already shown try: - while (message := self._read()) is not None: + while ( + message := self._read_relevant(thread=thread, ident=ident) + ) is not None: if message.get("id") == ident and "method" not in message: self._answer(message, said) + elif "id" in message and "method" not in message: + self._deferred.append(message) told: dict[str, Any] = message.get("params") or {} # One server holds every session of the agent, and a turn one of them # abandoned still says so on this stream. What is not this thread's is # not this turn's. if told.get("threadId") not in (None, thread): + self._deferred.append(message) continue named_turn: dict[str, Any] = told.get("turn") or {} if turn := told.get("turnId") or named_turn.get("id"): @@ -572,17 +673,10 @@ def turn(self, params: dict[str, Any], running: _Running) -> Iterator[Event]: # Sent as the turn spends it. `total` is the thread, every turn of # it; `last` is the one request that just came back. Cached input # is counted inside the input rather than beside it, so the input - # the server states is the whole of what went in -- and the two - # kinds together are the whole of what crossed the wire. + # the server states is the whole of what went in -- and `_usage` + # splits it once, so the cached part is not counted twice. counted: dict[str, Any] = told.get("tokenUsage") or {} - usage: dict[str, Any] = counted.get("total") or {} - held = Counter( - { - kind: int(usage.get(named) or 0) - for kind, named in _KINDS.items() - if usage.get(named) - } - ) + held = _usage(counted.get("total") or {}) if sum(held.values()): risen = Usage( { @@ -618,6 +712,9 @@ def turn(self, params: dict[str, Any], running: _Running) -> Iterator[Event]: # Codex reports a reconnect attempt as an error notification even # when a later sampling request completes this same turn. failed = None + running.last = str( + turn_said.get("id") or running.turn or "" + ) case "thread/status/changed" if ( begun and told["status"]["type"] == "idle" ): @@ -861,6 +958,38 @@ def _read(self, timeout: float | None = None) -> dict[str, Any] | None: Raises: subprocess.CalledProcessError: If the server stopped mid-turn. """ + if self._deferred: + return self._deferred.pop(0) + return self._read_raw(timeout) + + def _read_relevant( + self, + *, + thread: str, + ident: int, + timeout: float | None = None, + ) -> dict[str, Any] | None: + """Reads the next message for a thread/request, retaining all other messages.""" + + def relevant(message: dict[str, Any]) -> bool: + if "id" in message and "method" not in message: + return message.get("id") == ident + params = cast("dict[str, Any]", message.get("params") or {}) + item = cast("dict[str, Any]", params.get("item") or {}) + event_thread = params.get("threadId") or item.get("threadId") + return event_thread in (None, thread) + + for index, message in enumerate(self._deferred): + if relevant(message): + return self._deferred.pop(index) + message = self._read_raw(timeout) + while message is not None and not relevant(message): + self._deferred.append(message) + message = self._read_raw(timeout) + return message + + def _read_raw(self, timeout: float | None = None) -> dict[str, Any] | None: + """Reads only from the pump queue, bypassing deferred messages during routing.""" try: message = self._messages.get(timeout=timeout) except queue.Empty: @@ -872,6 +1001,19 @@ def _read(self, timeout: float | None = None) -> dict[str, Any] | None: ) return message + def _matching(self, predicate: Callable[[dict[str, Any]], bool]) -> dict[str, Any]: + """Waits for one message and preserves every message meant for another consumer.""" + for index, message in enumerate(self._deferred): + if predicate(message): + return self._deferred.pop(index) + while True: + message = self._read_raw() + if message is None: + raise Failed(self._proc.wait(), self._argv, "", "app server stopped") + if predicate(message): + return message + self._deferred.append(message) + def _answer(self, message: dict[str, Any], said: str) -> Any: """Unwraps one answer. @@ -966,6 +1108,16 @@ class CodexSession(SessionBase): #: reach this turn without a line being written into anybody's `config.toml`. takes_tools: ClassVar[bool] = True + #: `thread/fork` is Codex's native fork: it branches a thread in place, so the child + #: keeps the parent's prefix for the provider's cache. A fork is eager and prompt-free + #: here, and the child runs on a dedicated app server so parent and child may overlap. + forks: ClassVar[bool] = True + + @classmethod + def native_ready(cls) -> bool: + """Whether this installation's app-server schema exposes ``thread/fork``.""" + return _native_fork_ready(shutil.which("codex")) + def __init__( self, agent: AgentBase, cwd: str | os.PathLike[str] | None = None ) -> None: @@ -978,12 +1130,160 @@ def __init__( super().__init__(agent, cwd) #: The turn under way, which is what a word put in has to name. self._running = _Running() + #: The dedicated app server a forked child runs its turns on, or None for a session + #: nobody forked -- which runs on the agent's own, shared, server. + self._own: _AppServer | None = None @property def named(self) -> str | None: """The thread this session is, which the server names before the turn starts.""" return self._id or self._running.thread + def _server(self) -> _AppServer: + """The server this session's turns run on: its own for a forked child, else the agent's.""" + if self._fork_context is None: + return self._agent.server + if self._own is None: + context = self._fork_context + config = self._frozen_config + offering = tuple(sorted(one.name for one in self._toolbox().offered())) + own = _AppServer( + self._spawned( + self._agent._server_argv( + offering, + goals=context.goals, + web_search=context.web_search, + overrides=getattr(config, "overrides", ()), + toolbox=self._toolbox(), + ) + ), + self._environ(), + ) + own._held.append(weakref.ref(self._agent)) + self._own = own + weakref.finalize(self, _AppServer.stop, own) + return self._own + + def _model(self) -> str: + """The model this session runs at, off its fork context where it has one.""" + if self._fork_context is not None: + return self._fork_context.model + return self._agent.config.model + + def _effort_at(self) -> str: + """How hard this session's next turn thinks, off its fork context where it has one.""" + if self._fork_context is not None: + return self._fork_context.effort + return self.effort + + def _permission(self) -> str: + """The rung this session runs at, off its fork context where it has one.""" + if self._fork_context is not None: + return self._fork_context.permission + return self._agent.config.permission + + def _tier(self) -> str: + """The common provider tier this session runs at, off its fork context where it has one.""" + if self._fork_context is not None: + return self._fork_context.service_tier + return self._agent.config.service_tier + + def _fork(self, *, parent_id: str, last_turn_id: str | None) -> str: + """Performs Codex's native fork on a dedicated server, eagerly and prompt-free. + + The branch is made by ``thread/fork`` alone; the child's first turn goes through the + ordinary ``turn/start`` on this child server, using the effective values the fork + context froze. A non-None boundary names an earlier completed turn, inclusive. + + Args: + parent_id: The parent thread to branch. + last_turn_id: The completed turn to fork through, or None for the latest. + + Returns: + The child thread's id, which the child adopts. + """ + server = self._server() + params: dict[str, Any] = {"threadId": parent_id} + if last_turn_id is not None: + params["lastTurnId"] = last_turn_id + context = self._fork_context + if context is not None and context.permission_override: + params.update(server.permitted(context.permission, context.service_tier)) + try: + result = server.call("thread/fork", params, fallback=False) + result_map = cast( + "dict[str, Any]", result if isinstance(result, dict) else {} + ) + thread = result_map.get("thread") + thread_map = cast( + "dict[str, Any]", thread if isinstance(thread, dict) else {} + ) + child = thread_map.get("id") + if not isinstance(child, str) or not child: + raise Failed( + 1, server._argv, "", "thread/fork returned no child thread id" + ) + if self._fork_context is not None: + tier = result_map.get("serviceTier") + self._fork_context = replace( + self._fork_context, + model=str(result_map.get("model") or self._fork_context.model), + effort=str( + result_map.get("reasoningEffort") or self._fork_context.effort + ), + service_tier=( + "fast" + if tier == "priority" + else "default" + if tier == "default" + else self._fork_context.service_tier + ), + permission=_permission_from_fork( + result_map, self._fork_context.permission + ), + ) + return child # noqa: TRY300 -- malformed responses are handled below + except subprocess.CalledProcessError: + # The thread may have been created even though the response was lost. The fork is + # never retried -- that would create another branch -- so the orphan is written + # down for a person to reconcile, and the failure surfaces as it was. + if self._agent.cycle is not None: + self._agent.cycle.fork_lost( + self._agent, + parent_id, + last_turn_id, + provider=( + self._fork_context.provider + if self._fork_context is not None + else None + ), + ) + raise + except (TypeError, ValueError, AttributeError) as malformed: + if self._agent.cycle is not None: + self._agent.cycle.fork_lost( + self._agent, + parent_id, + last_turn_id, + provider=( + self._fork_context.provider + if self._fork_context is not None + else None + ), + ) + raise Failed(1, server._argv, "", str(malformed)) from malformed + + def _shut(self) -> None: + """Takes down only this child's dedicated server, never the parent agent's.""" + own, self._own = self._own, None + if own is not None: + own.stop() + + def _tools_changed(self) -> None: + """Restarts a fork server when its private tool list changes.""" + if self._fork_context is not None and self._own is not None: + self._shut() + def _stream( self, prompt: str, *, schema: type[BaseModel] | None = None ) -> Iterator[Event]: @@ -1003,7 +1303,7 @@ def _stream( with self._lock: # a conversation is a sequence: one turn at a time # Read once and held for the whole turn: asking the agent for its server again # may be starting another one, and the thread this turn is on is the first one's. - server = self._agent.server + server = self._server() thread = self._thread(server) # Known before the turn starts, so a word put in has a thread to name even though # the session is only opened once the turn has landed. The book goes with it: the @@ -1021,17 +1321,14 @@ def _stream( { "threadId": thread, "input": [{"type": "text", "text": prompt}], - "model": self._agent.config.model, - "effort": self.effort, + "model": self._model(), + "effort": self._effort_at(), **( {"outputSchema": schema.model_json_schema()} if schema is not None else {} ), - **server.permitted( - self._agent.config.permission, - self._agent.config.service_tier, - ), + **server.permitted(self._permission(), self._tier()), }, self._running, ): @@ -1050,6 +1347,9 @@ def _stream( # has had it already, as the turn said it. say(said, sys.stdout) self._adopt(thread) # a turn has landed, so the session is open + self._last_turn = self._running.last + if self._last_turn and self._last_turn not in self._completed_turns: + self._completed_turns = (*self._completed_turns, self._last_turn) yield Event(kind="result", text=said, tokens=spent, spent=costing) def interject(self, text: str) -> None: @@ -1087,16 +1387,14 @@ def _thread(self, server: _AppServer) -> str: Returns: The thread's id, which is also the session's. """ - rung = server.permitted( - self._agent.config.permission, self._agent.config.service_tier - ) + rung = server.permitted(self._permission(), self._tier()) if (thread := self._id) is None: return str( server.call( "thread/start", { "cwd": self._workspace(), - "model": self._agent.config.model, + "model": self._model(), **rung, }, )["thread"]["id"] @@ -1120,17 +1418,16 @@ def _pursue(self, objective: str) -> str: leaving the session unopened so that the next call retries it. """ with self._lock: # a conversation is a sequence: one turn at a time - server = self._agent.server - config = self._agent.config + server = self._server() thread = self._thread(server) server.call("thread/goal/set", {"threadId": thread, "objective": objective}) answer = server.pursue( { "threadId": thread, "input": [{"type": "text", "text": objective}], - "model": config.model, - "effort": self.effort, - **server.permitted(config.permission, config.service_tier), + "model": self._model(), + "effort": self._effort_at(), + **server.permitted(self._permission(), self._tier()), } ) self._adopt(thread) @@ -1215,42 +1512,14 @@ def server(self) -> _AppServer: # is stopped by its own finalizer when the agent is collected either way. self._server, self._server_as = None, "" if self._server is None: - argv = ["codex", "app-server"] - if not self.goals_enabled: - # Per server rather than in config, so this flow changes no other Codex - # session belonging to the user. - argv += ["--disable", "goals"] - # Said in both directions rather than only when it is off: Codex searches - # nothing until it is asked to, so an agent that may search the web has to - # say so here for `web_search` to mean on every backend what it says. - argv += [ - "-c", - f"tools.web_search={'true' if self.config.web_search else 'false'}", - ] - argv += ["--stdio"] - for key, value in getattr(self.config, "overrides", ()): - # The same `-c` Codex's own client takes, scoped to this server: a - # window asked for here is this agent's, and the user's config.toml is - # left exactly as it was. - argv += ["-c", f"{key}={value}"] - if offering: - # The flow's own callbacks, as the one thing Codex takes a tool it was - # not shipped with on. Scoped to this server for the reason the overrides - # are: nothing of the user's `config.toml` is written, and no other Codex - # they are running is told about a tool that belongs to this flow. - held = self.toolbox.command() - argv += [ - "-c", - f"mcp_servers.humanize.command={json.dumps(held[0])}", - "-c", - f"mcp_servers.humanize.args={json.dumps(held[1:])}", - ] # Read before the environment is built out of it: a fallback landing # between the two reads would name the account this server is *not* signed # into, and a server that believes it is already elsewhere is one nothing ever # starts again. account = self.node().name - self._server = _AppServer(self.spawned(argv), self._environ()) + self._server = _AppServer( + self.spawned(self._server_argv(offering)), self._environ() + ) self._server_as, self._server_tools = account, offering self._server._held.append(weakref.ref(self)) # Held by the finalizer alone, which is what takes the server down: when the @@ -1258,6 +1527,86 @@ def server(self) -> _AppServer: weakref.finalize(self, self._server.stop) return self._server + def _server_argv( + self, + offering: tuple[str, ...], + *, + goals: bool | object = _LIVE, + web_search: bool | object = _LIVE, + overrides: Sequence[tuple[str, str]] | object = _LIVE, + toolbox: Toolbox | object = _LIVE, + ) -> list[str]: + """The command that starts an app server for this agent, given what it offers. + + Args: + offering: The flow's callbacks this server is to be told about, by name. + goals: Whether backend goals are enabled, or the live agent setting when omitted. + web_search: Whether web search is enabled, or the live agent setting when omitted. + overrides: Codex process overrides, or the live config when omitted. + toolbox: The callback bridge to expose, or the live agent toolbox when omitted. + + Returns: + The argv, over the same home and MCP bridge the agent's own server uses. + """ + selected_goals = self.goals_enabled if goals is _LIVE else bool(goals) + selected_search = ( + self.config.web_search if web_search is _LIVE else bool(web_search) + ) + selected_overrides: tuple[tuple[str, str], ...] = ( + tuple(getattr(self.config, "overrides", ())) + if overrides is _LIVE + else tuple(cast("Sequence[tuple[str, str]]", overrides)) + ) + selected_toolbox = self.toolbox if toolbox is _LIVE else toolbox + argv = ["codex", "app-server"] + if not selected_goals: + # Per server rather than in config, so this flow changes no other Codex + # session belonging to the user. + argv += ["--disable", "goals"] + # Said in both directions rather than only when it is off: Codex searches + # nothing until it is asked to, so an agent that may search the web has to + # say so here for `web_search` to mean on every backend what it says. + argv += [ + "-c", + f"tools.web_search={'true' if selected_search else 'false'}", + ] + argv += ["--stdio"] + for key, value in selected_overrides: + # The same `-c` Codex's own client takes, scoped to this server: a + # window asked for here is this agent's, and the user's config.toml is + # left exactly as it was. + argv += ["-c", f"{key}={value}"] + if offering: + # The flow's own callbacks, as the one thing Codex takes a tool it was + # not shipped with on. Scoped to this server for the reason the overrides + # are: nothing of the user's `config.toml` is written, and no other Codex + # they are running is told about a tool that belongs to this flow. + assert isinstance(selected_toolbox, Toolbox) # noqa: S101 + held = selected_toolbox.command() + argv += [ + "-c", + f"mcp_servers.humanize.command={json.dumps(held[0])}", + "-c", + f"mcp_servers.humanize.args={json.dumps(held[1:])}", + ] + return argv + + def spawn_server(self) -> _AppServer: + """A fresh app server for a forked child, over the same home, machine and bridge. + + The child runs its turns here rather than on the agent's server, so that a child turn + may overlap a parent turn: the agent's own server stays serialized over its sessions, + and the fork runtime owns this separate one. Held by the child, which stops it alone -- + never the parent agent's server. + + Returns: + The server, already introduced to. + """ + offering = tuple(sorted(one.name for one in self.toolbox.offered())) + server = _AppServer(self.spawned(self._server_argv(offering)), self._environ()) + server._held.append(weakref.ref(self)) + return server + def stop(self) -> None: """Takes no further turn, and takes down the server the turn under way is waiting on.""" super().stop() diff --git a/src/hmz/agents/config.py b/src/hmz/agents/config.py index 9985652c..80d8d020 100644 --- a/src/hmz/agents/config.py +++ b/src/hmz/agents/config.py @@ -16,6 +16,7 @@ "SERVICE_TIERS", "AgentConfig", "AgentDefaults", + "Forks", "Goal", "Isolated", "Remote", @@ -63,6 +64,22 @@ class Agents(NamedTuple): """ +class Forks: + """What a flow writes beside an agent it will branch a conversation of. + + `Session.fork` branches an already-open conversation into an independent one, preserving + the parent's prefix for the backend's own cache. Only Claude and Codex have a native + history operation for it, so a flow built on it is not a flow any agent can drive. It says + which of its agents has to have one, by writing this where it declares them:: + + class Agents(NamedTuple): + worker: Annotated[AgentBase, Forks] + + and an agent whose backend has no native fork is refused before the first turn rather than + raising in the middle of one, which is where a loop would otherwise find out. + """ + + @dataclass(frozen=True, slots=True, kw_only=True) class AgentDefaults: """The initial goal availability offered for one place in a flow. diff --git a/src/hmz/cycle.py b/src/hmz/cycle.py index ab8742ce..19457239 100644 --- a/src/hmz/cycle.py +++ b/src/hmz/cycle.py @@ -54,6 +54,7 @@ from collections.abc import Mapping, Sequence from .agents import AgentBase + from .agents.event import Usage from .tracing.profile import Profiler __all__ = [ @@ -73,6 +74,7 @@ "Sub", "called", "cycles", + "forks", "linked", "opened", "read", @@ -828,6 +830,108 @@ def opened(self, agent: AgentBase, session: str) -> None: ) self.links(name) + def forked( + self, + agent: AgentBase, + parent: str, + child: str, + last_turn_id: str | None = None, + *, + provider: str | None = None, + permission: str | None = None, + cache_equivalent: bool = True, + ) -> None: + """Writes down that one session branched into another, at a completed boundary. + + A fork is not an open: the child did not start from nothing, so the run records the + parent it came from and the boundary it came off. Both ids are written for + diagnostics, and both relation keys -- the cycle's own names for the two sessions -- + for whoever links the trace afterwards. + + Args: + agent: Whose conversation it is, which is the same agent for both. + parent: The parent's backend id. + child: The child's backend id, just given by the native fork. + last_turn_id: The completed turn the child branched from, or None for a backend + whose fork takes no intermediate boundary. + provider: The effective provider snapshot at the fork boundary, if already known. + permission: The child's effective permission. + cache_equivalent: Whether the child retains the parent's cache-equivalent settings. + """ + account = _provider(agent) if provider is None else provider + parent_name = called(agent.id, agent.backend, account, parent) + child_name = called(agent.id, agent.backend, account, child) + with self._writing: + self._sessions[child_name] = (agent.backend, child) + self.write( + "forked", + agent=agent.id, + backend=agent.backend, + provider=account or LOCAL, + parent_session_id=parent, + session_id=child, + parent_key=parent_name, + session_key=child_name, + permission=permission, + cache_equivalent=cache_equivalent, + **({"last_turn_id": last_turn_id} if last_turn_id else {}), + ) + self.links(child_name) + + def fork_usage(self, agent: AgentBase, session: str, usage: Usage) -> None: + """Writes numeric usage for one completed fork child turn.""" + if not usage.total: + return + self.write( + "fork-usage", + agent=agent.id, + backend=agent.backend, + session_id=session, + **dict(usage), + total=usage.total, + ) + + def fork_failed(self, agent: AgentBase, session: str, error: str) -> None: + """Writes a bounded diagnostic for a fork child turn that failed.""" + self.write( + "fork-failed", + agent=agent.id, + backend=agent.backend, + session_id=session, + error=" ".join(error.split())[:400], + ) + + def fork_lost( + self, + agent: AgentBase, + parent: str, + last_turn_id: str | None = None, + *, + provider: str | None = None, + ) -> None: + """Writes down a fork whose child id was lost with the response that made it. + + A native fork that fails after the backend may already have created the child is not + retried -- a retry would create another branch -- so the orphan is written down for a + person to reconcile rather than left to multiply in silence. + + Args: + agent: Whose conversation it is. + parent: The parent's backend id, which the branch was made from. + last_turn_id: The boundary the branch was made at, or None where the backend takes + none. + provider: The effective provider snapshot, or None to resolve it from the agent. + """ + provider = _provider(agent) if provider is None else provider + self.write( + "fork-lost", + agent=agent.id, + backend=agent.backend, + provider=provider or LOCAL, + parent_session_id=parent, + **({"last_turn_id": last_turn_id} if last_turn_id else {}), + ) + def links(self, only: str = "") -> None: """Points this cycle's `sessions/` at the logs its sessions are being written to. @@ -996,6 +1100,31 @@ def records(cycle: Path) -> list[Path]: return held +def forks(cycle: Path) -> dict[str, str]: + """What each session one cycle branched from, child id to parent id. + + A forked child did not start from nothing, so a trace of the run draws it as the child of + the conversation it branched from. Read across every record of the cycle, as + :func:`opened` is, and keyed by the backend ids -- which is what a trace is gathered by. + + Args: + cycle: The cycle to read. + + Returns: + One entry per fork, the child's id to the parent's id. Empty for a run that forked + nothing. + """ + held: dict[str, str] = {} + for at in records(cycle): + for said in _events(at): + if said.get("event") != "forked" or not said.get("session_id"): + continue + parent = said.get("parent_session_id") + if isinstance(parent, str): + held[str(said["session_id"])] = parent + return held + + def opened(cycle: Path) -> dict[str, list[str]]: """What each agent of one cycle opened, as the ids the backends gave those sessions. @@ -1042,13 +1171,18 @@ def sessions(cycle: Path) -> list[Session]: "", ) for said in events: - if said.get("event") != "opened" or not said.get("session"): + if said.get("event") == "opened" and said.get("session"): + ident = str(said["session"]) + elif said.get("event") == "forked" and said.get("session_id"): + # A forked child is a session the run opened, even though it did not start + # from nothing: it is read back the same way, keyed by the id the fork gave. + ident = str(said["session_id"]) + else: continue agent, backend = ( str(said.get("agent") or ""), str(said.get("backend") or ""), ) - ident = str(said["session"]) provider = str(said.get("provider") or LOCAL) held.append( Session( @@ -1058,9 +1192,11 @@ def sessions(cycle: Path) -> list[Session]: ident=ident, # Worked out where an older cycle did not write one down: a name is what # this session is called, and a cycle written before it had one still - # has sessions. + # has sessions. A fork writes its own name, as `session_key`. name=str( - said.get("name") or called(agent, backend, provider, ident) + said.get("name") + or said.get("session_key") + or called(agent, backend, provider, ident) ), at=str(said.get("at") or ""), flow=flow, diff --git a/src/hmz/flows/__init__.py b/src/hmz/flows/__init__.py index ef848119..3e04372f 100644 --- a/src/hmz/flows/__init__.py +++ b/src/hmz/flows/__init__.py @@ -108,6 +108,7 @@ def run(agents: tuple[Agent, Agent], task: str) -> None: Board, Event, Failed, + Forks, Goal, Hook, Hooks, @@ -170,6 +171,7 @@ def run(agents: tuple[Agent, Agent], task: str) -> None: "Finding", "Flow", "Flowverse", + "Forks", "Goal", "Hook", "Hooks", @@ -282,6 +284,7 @@ def run(agents: tuple[Agent, Agent], task: str) -> None: "EVERYWHERE": "hmz.agents", "Event": "hmz.agents", "Failed": "hmz.agents", + "Forks": "hmz.agents", "Goal": "hmz.agents", "Hook": "hmz.agents", "Hooks": "hmz.agents", diff --git a/src/hmz/flows/agent.py b/src/hmz/flows/agent.py index cfebf954..7e3eaed5 100644 --- a/src/hmz/flows/agent.py +++ b/src/hmz/flows/agent.py @@ -67,6 +67,12 @@ class Session(Protocol): #: to catch the refusal. takes_tools: ClassVar[bool] + #: Whether this backend has a native history operation that branches a conversation in + #: place, which is what :meth:`fork` reaches for. Only Claude and Codex do; a flow that + #: forks declares the place with `Annotated[Agent, Forks]`, and the run is refused a + #: backend without it before the first turn. + forks: ClassVar[bool] + @property def id(self) -> str: """What the backend calls this conversation, once a turn has landed in it. @@ -81,6 +87,35 @@ def named(self) -> str | None: """The same name, or None while the backend has not said one.""" ... + @property + def last_turn_id(self) -> str | None: + """The backend's id for the latest completed turn, where it exposes one. + + What a fork names its boundary by; None for a backend with no intermediate boundary + and before any turn has completed. A forked child starts empty here. + """ + ... + + def fork( + self, *, last_turn_id: str | None = None, permission: str | None = None + ) -> Session: + """Branches this conversation into an independent one, preserving its prefix. + + Eager and prompt-free: the returned child already has its backend id, and the parent + is left open, idle and unchanged, so it may be driven on at once while the child runs + on its own. Only an open, idle, unmoved session may be forked. + + Args: + last_turn_id: The completed turn to fork through, inclusive. None forks through the + latest completed turn; Codex also accepts an earlier one, and Claude raises + NotImplementedError for a non-None boundary. + permission: The rung the child runs at, or None to inherit the parent's. + + Returns: + The child session, already named by the backend. + """ + ... + @property def cwd(self) -> str: """The directory this conversation works in, as whoever is watching would name it.""" diff --git a/src/hmz/flows/checking.py b/src/hmz/flows/checking.py index 30b46ab9..4ac35e31 100644 --- a/src/hmz/flows/checking.py +++ b/src/hmz/flows/checking.py @@ -1429,10 +1429,10 @@ def _valued(value: ast.expr, scope: _Scope) -> frozenset[str] | _Crew | _Answer def _called(value: ast.Call, scope: _Scope) -> frozenset[str] | _Crew | _Answer | None: """What calling one thing answers with, where what is called is tracked.""" func = value.func - opened = isinstance(func, ast.Attribute) and func.attr in {"new", "clone"} + opened = isinstance(func, ast.Attribute) and func.attr in {"new", "clone", "fork"} target = func.value if isinstance(func, ast.Attribute) and opened else func held = _valued(target, scope) if opened else None - if isinstance(func, ast.Attribute) and func.attr == "new": + if isinstance(func, ast.Attribute) and func.attr in {"new", "fork"}: if isinstance(held, frozenset) and held: return frozenset({"session"}) return None @@ -2177,6 +2177,19 @@ def catalogue() -> tuple[Capability, ...]: "and is refused an agent whose backend has none before the first turn", ) ) + forking = frozenset( + name for name, one in sessions.items() if getattr(one, "forks", False) + ) + held.append( + Capability( + "forks", + forking, + "branch a conversation in place, preserving its prefix -- " + "child = session.fork() answers a session already named by the backend, and " + "the parent is left open and unchanged -- which a backend not among these " + "refuses; a flow that forks declares the place, Annotated[Agent, Forks]", + ) + ) return tuple(held) diff --git a/src/hmz/flows/driving.py b/src/hmz/flows/driving.py index 05b6482c..56024f54 100644 --- a/src/hmz/flows/driving.py +++ b/src/hmz/flows/driving.py @@ -331,6 +331,9 @@ class Place(NamedTuple): goal: Whether the flow runs this one under the backend's own goal feature, which it said by writing `Annotated[Agent, Goal]` where it declared the place. Only four backends have one, so a flow built on it is not a flow any agent can drive. + forks: Whether the flow branches a conversation of this one, which it said by writing + `Annotated[Agent, Forks]` where it declared the place. Only Claude and Codex have a + native fork, so a flow built on it is not a flow any agent can drive. goals_default: Whether the agent picker initially offers backend goals on or off for this place, which a flow may suggest with `AgentDefaults(goals=False)`. Once selected, the effective value belongs to the agent's config. A required `Goal` always starts on. @@ -348,6 +351,7 @@ class Place(NamedTuple): where: type[Remote] | Remote | Isolated | None = None goal: bool = False goals_default: bool = True + forks: bool = False def drives(flow: str | os.PathLike[str]) -> tuple[str, ...]: @@ -1317,6 +1321,7 @@ def _place(name: str, kind: object) -> Place: where = _where(kind) goal = _goal(kind) goals_default = _goals_default(kind) + forks = _forks(kind) if get_origin(kind) is Annotated: kind = get_args(kind)[0] return Place( @@ -1326,6 +1331,7 @@ def _place(name: str, kind: object) -> Place: where=where, goal=goal, goals_default=True if goal else goals_default, + forks=forks, ) @@ -1366,6 +1372,23 @@ def _goal(kind: object) -> bool: return any(said is Goal for said in get_args(kind)[1:]) +def _forks(kind: object) -> bool: + """Whether a flow said it branches a conversation of the agent filling a place. + + Args: + kind: What the flow annotated the place with. + + Returns: + True if it wrote `Forks` beside the type, and False for a place annotated with the type + alone -- which is one driven by turns like every other. + """ + from hmz.agents import Forks + + if get_origin(kind) is not Annotated: + return False + return any(said is Forks for said in get_args(kind)[1:]) + + def _goals_default(kind: object) -> bool: """The initial on/off choice a flow suggests for this agent's goals. diff --git a/src/hmz/runner.py b/src/hmz/runner.py index c1cfe2a0..183675bf 100644 --- a/src/hmz/runner.py +++ b/src/hmz/runner.py @@ -30,6 +30,38 @@ __all__ = ["Runner", "flow_and_agents", "read_agent", "set_up_from"] +def _forkable(agent: AgentBase) -> bool: + """Whether this agent's backend has a native fork, read off the session it opens. + + `forks` is a fact of the session rather than of the agent, the way `shapes` and + `takes_tools` are -- the class `new` answers with is what carries it. Resolved by name + rather than through `get_type_hints`, which would ask every annotation to resolve. + + Args: + agent: The agent whose backend is being asked. + + Returns: + True for a backend whose session declares `forks`, False otherwise. + """ + import contextlib + import inspect + import sys as running + + told: object = None + with contextlib.suppress(Exception): + told = inspect.signature(type(agent).new).return_annotation + if isinstance(told, str): + told = vars(running.modules[type(agent).__module__]).get(told) + if not isinstance(told, type) or not getattr(told, "forks", False): + return False + ready = getattr(told, "native_ready", None) + if not callable(ready): + return True + with contextlib.suppress(Exception): + return bool(ready()) + return False + + def _finished(running: Awaitable[None]) -> None: """Runs a flow that is a coroutine, until it returns. @@ -146,6 +178,11 @@ def __init__( f"{flow}: {place.name or 'the agent'} is run under a goal, but goals " "were switched off for it" ) + if place.forks and not _forkable(agent): + raise NotAFlow( + f"{flow}: {place.name or 'the agent'} branches a conversation, which " + f"{agent.backend} has no native fork for" + ) lands(flow, agent, place) # The person at the prompt is made here rather than given: nobody chooses what they # run, so nothing upstream of this was ever asked about them. diff --git a/src/hmz/sdk/cycles.py b/src/hmz/sdk/cycles.py index af70999a..ac49ee4d 100644 --- a/src/hmz/sdk/cycles.py +++ b/src/hmz/sdk/cycles.py @@ -106,7 +106,7 @@ def traced( """ import datetime - from hmz.cycle import TRACES + from hmz.cycle import TRACES, forks from hmz.tracing.profile import PROFILE agents = self.opened(cycle) @@ -123,6 +123,7 @@ def traced( start=start, end=end, profile=cycle / PROFILE, + parents=forks(cycle), ) return where, document @@ -135,6 +136,7 @@ def trace( start: str | None = None, end: str | None = None, profile: str | os.PathLike[str] | None = None, + parents: Mapping[str, str] | None = None, ) -> dict[str, Any]: """Gathers what a run left behind into one Chrome trace. @@ -148,6 +150,8 @@ def trace( start: The earliest session time to include, in any wording dateparser understands. end: The latest. profile: Where the run's own profile was written, for a run that was profiled. + parents: What each session branched from, child id to parent id, for the forked + children a trace draws under their parent. Returns: The trace, as the object that was written. @@ -162,4 +166,5 @@ def trace( start=start, end=end, profile=profile, + parents=parents, ) diff --git a/src/hmz/tracing/collector.py b/src/hmz/tracing/collector.py index dec9790b..4f046ea4 100644 --- a/src/hmz/tracing/collector.py +++ b/src/hmz/tracing/collector.py @@ -66,6 +66,7 @@ def collect( start: str | None = None, end: str | None = None, profile: str | os.PathLike[str] | Iterable[Process] | None = None, + parents: Mapping[str, str] | None = None, ) -> dict[str, Any]: """Aggregates agent trajectories into a Chrome trace. @@ -103,6 +104,11 @@ def collect( the trace with a track per thread, beside the agents' own: a turn is mostly other programs, and one timeline is what makes that visible. + parents: What each session branched from, child id to parent id, which + is what a run's `forked` events record. A forked child is drawn as + the child of the conversation it came from, where nothing in its + own log says it did -- a fork is a native history operation, not a + sub-agent. Returns: The Chrome trace document, also written to output when one is given. @@ -154,6 +160,12 @@ def collect( if reader is not None and home.is_dir(): collected += reader(home, root, names, window) + if parents: + for item in collected: + parent = parents.get(item.ident) + if parent is not None: + # A fork stays on one backend, so the parent's key carries the child's. + item.parent = f"{item.backend}:{parent}" named = {ident: name for name, opened in (agents or {}).items() for ident in opened} known = {item.key: item for item in collected} for item in collected: diff --git a/tests/agents/test_appservers.py b/tests/agents/test_appservers.py index 44e8a0b0..5b32f59d 100644 --- a/tests/agents/test_appservers.py +++ b/tests/agents/test_appservers.py @@ -768,6 +768,25 @@ def test_what_a_codex_turn_spent_is_charged_to_the_turn_that_spent_it( assert second[-1].tokens == {"gpt-5-codex": 500} # 1500 all told, 1000 of it before +def test_a_codex_turn_counts_cached_input_once_as_cache_read( + working: _FakeServer, +) -> None: + """The server states the input with the cached part inside it, so it is split once.""" + session = CodexAgent(CodexAgentConfig(model="gpt-5-codex", effort="high")).new() + + (result,) = [ + event for event in session.stream("do the task") if event.kind == "result" + ] + + # Net-new input, cached input and output: the three together are the whole of what + # crossed the wire, and the cached part is not counted twice. + assert result.spent["cache_read"] == 800 + assert result.spent["input"] == 100 # net-new, not the whole 900 + assert result.spent["output"] == 100 + assert result.spent.total == 1000 + assert result.tokens == {"gpt-5-codex": 1000} + + def test_a_codex_session_with_no_turn_running_cannot_be_talked_to() -> None: session = CodexAgent(CodexAgentConfig(model="gpt-5-codex", effort="high")).new() diff --git a/tests/agents/test_forking.py b/tests/agents/test_forking.py new file mode 100644 index 00000000..2fe51e97 --- /dev/null +++ b/tests/agents/test_forking.py @@ -0,0 +1,777 @@ +"""Branching a conversation in place, preserving its prefix for the backend's cache. + +Only Claude and Codex have a native history operation for it -- `--fork-session` and +`thread/fork` -- so a flow built on `Session.fork` declares the place with `Annotated[Agent, +Forks]` and is refused an unfit backend before the first turn. Both drivers are exercised +against stand-ins, so what is checked is the exact call a fork is made of, the boundary it +honours, and the isolation of the child from whatever the parent becomes afterwards. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import pytest + +from hmz.agents import ( + ClaudeCodeAgent, + ClaudeCodeAgentConfig, + ClaudeCodeSession, + CodexAgent, + CodexAgentConfig, + CodexSession, + OpencodeAgent, + OpencodeAgentConfig, + PiAgent, + PiAgentConfig, +) +from hmz.flows import NotAFlow, wanted +from hmz.runner import Runner + +#: A `claude --print`: it names the session it was given, forks without a prompt when asked, +#: and otherwise answers each `user` message written to it as one turn. A prompt of `boom` is +#: refused, which is how a child whose first turn fails is spelled. +_CLAUDE = """ +import json, pathlib, sys + +LOG = pathlib.Path(sys.argv[0] + ".log") + + +def note(entry): + with LOG.open("a") as stream: + json.dump(entry, stream) + stream.write("\\n") + + +def out(said): + print(json.dumps(said), flush=True) + + +argv = sys.argv[1:] +note({"argv": argv}) +flags = {} +for i, one in enumerate(argv): + if one.startswith("--") and i + 1 < len(argv) and not argv[i + 1].startswith("--"): + flags[one] = argv[i + 1] + elif one.startswith("--"): + flags[one] = True +sid = flags.get("--session-id") or flags.get("--resume") +out({"type": "system", "session_id": sid}) +if flags.get("--fork-session"): + sys.exit(0) # the fork is made by the flags alone; no prompt is owed +for line in sys.stdin: + said = json.loads(line) + if said.get("type") != "user": + continue + text = said["message"]["content"][0]["text"] + if text == "boom": + out({"type": "result", "subtype": "error_during_execution", "is_error": True, + "result": ""}) + continue + out({"type": "assistant", "message": {"id": "msg_1", "role": "assistant", + "content": [{"type": "text", "text": text}]}}) + out({"type": "result", "subtype": "success", "is_error": False, "result": text}) +""" + +#: A `codex app-server`: it answers `thread/start` with one thread, `thread/fork` with another, +#: and completes whatever turn is started on either. Both the parent's server and the forked +#: child's dedicated server run this same stand-in, so what a test reads is the calls each was +#: made of and how many servers there were. +_CODEX = """ +import json, pathlib, sys + +LOG = pathlib.Path(sys.argv[0] + ".log") + + +def send(message): + print(json.dumps(message), flush=True) + + +for line in sys.stdin: + call = json.loads(line) + with LOG.open("a") as stream: + json.dump(call, stream) + stream.write("\\n") + if "id" not in call: + continue + method = call.get("method") + result = {} + if method == "thread/start": + result = {"thread": {"id": "parent_thread"}} + elif method == "thread/fork": + result = {"thread": {"id": "child_thread"}} + send({"method": "thread/status/changed", + "params": {"threadId": "other_thread", "status": {"type": "idle"}}}) + send({"jsonrpc": "2.0", "id": call["id"], "result": result}) + if method == "thread/start": + send({"method": "thread/status/changed", + "params": {"status": {"type": "idle"}}}) + if method == "turn/start": + tid = call["params"].get("threadId", "parent_thread") + send({"method": "turn/started", + "params": {"turnId": "turn_fake", "threadId": tid}}) + send({"method": "item/completed", + "params": {"item": {"type": "agentMessage", "text": "answered"}, + "threadId": tid}}) + send({"method": "turn/completed", + "params": {"threadId": tid, + "turn": {"id": "turn_fake", "status": "completed"}}}) + send({"method": "thread/status/changed", + "params": {"status": {"type": "idle"}, "threadId": tid}}) +""" + +#: A flow that branches a conversation, which only some backends can. +FORKING = '''"""A loop that hands the work to a forked child rather than the agent itself.""" + +from typing import Annotated, NamedTuple + +from hmz.agents import AgentBase, Forks +from hmz.flows import flow + + +class Agents(NamedTuple): + """The one it drives, which has to have a native fork.""" + + worker: Annotated[AgentBase, Forks] + + +@flow +def run(agents: Agents, task: str) -> None: + agents.worker.new().fork() +''' + + +@dataclass(frozen=True) +class _Fake: + """A stand-in backend on PATH, and everything it was asked for.""" + + log: Path + + def calls(self) -> list[dict[str, Any]]: + return [json.loads(line) for line in self.log.read_text().splitlines()] + + +def _install( + name: str, script: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> _Fake: + """Puts one stand-in CLI on PATH under the name the backend calls it.""" + binaries = tmp_path / "bin" + binaries.mkdir(exist_ok=True) + fake = binaries / name + fake.write_text(f"#!{sys.executable}\n{script}") + fake.chmod(0o755) + monkeypatch.setenv("PATH", f"{binaries}{os.pathsep}{os.environ['PATH']}") + return _Fake(Path(f"{fake}.log")) + + +@pytest.fixture +def claude(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> _Fake: + return _install("claude", _CLAUDE, tmp_path, monkeypatch) + + +@pytest.fixture +def codex(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> _Fake: + return _install("codex", _CODEX, tmp_path, monkeypatch) + + +def _written(tmp_path: Path, source: str, name: str = "forking") -> str: + """Writes a flow out and answers with its path.""" + where = tmp_path / f"{name}.py" + where.write_text(source) + return str(where) + + +# --- The capability a fork is built on ------------------------------------------------------- + + +def test_only_claude_and_codex_say_a_fork_is_available() -> None: + """The one thing that gates the `Forks` declaration, read off the session class.""" + assert ClaudeCodeSession.forks is True + assert CodexSession.forks is True + assert ( + OpencodeAgent(OpencodeAgentConfig(model="m", effort="low")).new().forks is False + ) + + +def test_an_unsupported_backend_refuses_a_fork_before_a_child_is_made() -> None: + """No child is created and nothing is sent to the backend: the refusal is up front.""" + for agent in ( + PiAgent(PiAgentConfig(model="m", effort="low")), + OpencodeAgent(OpencodeAgentConfig(model="m", effort="low")), + ): + with pytest.raises(NotImplementedError, match="no native fork"): + agent.new().fork() + + +def test_a_fork_needs_an_opened_parent() -> None: + """Eager and prompt-free, but only once there is a conversation to branch.""" + session = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")).new() + with pytest.raises(RuntimeError, match="has not run a turn yet"): + session.fork() + + +def test_a_fork_refuses_a_parent_whose_turn_is_still_running() -> None: + """A fork while a turn is running would branch a conversation mid-answer.""" + session = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")).new() + session._id = "parent" + session._working = True + with pytest.raises(RuntimeError, match="while a turn is running"): + session.fork() + + +def test_a_fork_refuses_a_closed_parent() -> None: + session = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")).new() + session._id = "parent" + session._ended = True + with pytest.raises(RuntimeError, match="closed"): + session.fork() + + +def test_a_fork_refuses_a_parent_that_moved_backends() -> None: + agent = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")) + session = agent.new() + session._id = "parent" + session._moved_to = agent.new() + with pytest.raises(RuntimeError, match="moved"): + session.fork() + + +def test_a_fork_refuses_an_unknown_permission(claude: _Fake) -> None: + agent = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")) + session = agent.new() + assert session("hi") == "hi" + with pytest.raises(ValueError, match="permission must be one of"): + session.fork(permission="everything") + + +# --- The Forks declaration, checked before the first turn ------------------------------------ + + +def test_a_place_that_forks_says_so(tmp_path: Path) -> None: + (place,) = wanted(_written(tmp_path, FORKING)) + + assert place.forks is True + assert place.name == "worker" + + +def test_an_agent_without_a_native_fork_is_refused(tmp_path: Path) -> None: + where = _written(tmp_path, FORKING) + + with pytest.raises(NotAFlow, match="has no native fork for"): + Runner(where, [PiAgent(PiAgentConfig(model="m", effort="low"))]) + + +@pytest.mark.parametrize( + "agent", + [ + ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")), + CodexAgent(CodexAgentConfig(model="m", effort="low")), + ], +) +def test_an_agent_whose_backend_forks_is_taken(agent: object, tmp_path: Path) -> None: + runner = Runner(_written(tmp_path, FORKING), [agent]) # pyright: ignore[reportArgumentType] + + assert len(runner.agents) == 1 + + +def test_the_catalogue_says_which_backends_fork() -> None: + from hmz.flows import catalogue + + (forks,) = [one for one in catalogue() if one.name == "forks"] + + assert forks.backends == frozenset({"claude", "codex"}) + + +# --- Claude: `--resume --fork-session --session-id` ------------------------------------------ + + +def test_claude_forks_eagerly_and_the_child_resumes_itself(claude: _Fake) -> None: + """The branch is made by the flags alone, and the child carries on under its own id.""" + agent = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")) + parent = agent.new() + assert parent("hi") == "hi" + + child = parent.fork() + + assert child("research") == "research" + assert parent("next") == "next" # the parent goes on, unchanged + + calls = claude.calls() + forked = next(call for call in calls if "--fork-session" in call["argv"]) + assert forked["argv"][forked["argv"].index("--resume") + 1] == parent.id + assert forked["argv"][forked["argv"].index("--session-id") + 1] == child.id + # The child's own turn resumes the child, never the parent. + child_opened = next( + call + for call in calls + if "--resume" in call["argv"] + and call["argv"][call["argv"].index("--resume") + 1] == child.id + ) + assert "--fork-session" not in child_opened["argv"] + assert parent.id != child.id + + +def test_claude_refuses_an_intermediate_boundary(claude: _Fake) -> None: + """Claude forks the whole conversation, so a non-None boundary is not silently ignored.""" + session = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")).new() + assert session("hi") == "hi" + + with pytest.raises(NotImplementedError, match="whole conversation"): + session.fork(last_turn_id="some-turn") + + +def test_a_claude_child_does_not_see_tools_added_to_the_parent_after_fork( + claude: _Fake, +) -> None: + """The child bridge is private, so a later parent offer cannot leak into its turn.""" + from hmz.agents import Tool + + agent = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")) + parent = agent.new() + assert parent("hi") == "hi" + child = parent.fork() + parent.offers([Tool(name="later", about="a later callback", call=lambda: "later")]) + child.offers([]) + + assert child("research") == "research" + child_call = next( + call + for call in claude.calls() + if "--resume" in call["argv"] + and call["argv"][call["argv"].index("--resume") + 1] == child.id + ) + assert "--mcp-config" not in child_call["argv"] + + +def test_a_claude_child_whose_first_turn_fails_is_still_the_child( + claude: _Fake, +) -> None: + """The child id was adopted by the fork, so a failed turn is its own failure, not the fork.""" + session = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")).new() + assert session("hi") == "hi" + child = session.fork() + + with pytest.raises(subprocess.CalledProcessError): + child("boom") + + assert child.id != session.id # the child exists even though the turn failed + + +def test_reconfiguring_the_parent_does_not_change_a_claude_child(claude: _Fake) -> None: + """The fork context froze the boundary: a later reconfiguration is the parent's own.""" + from dataclasses import replace + + agent = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")) + parent = agent.new() + assert parent("hi") == "hi" + child = parent.fork() + + agent.reconfigure(replace(agent.config, model="changed", effort="max")) + assert child("research") == "research" + + child_opened = next( + call + for call in claude.calls() + if "--resume" in call["argv"] + and call["argv"][call["argv"].index("--resume") + 1] == child.id + ) + assert child_opened["argv"][child_opened["argv"].index("--model") + 1] == "m" + assert child_opened["argv"][child_opened["argv"].index("--effort") + 1] == "low" + + +def test_a_fork_child_does_not_copy_pending_agent_waiting_prompts( + claude: _Fake, +) -> None: + agent = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")) + parent = agent.new() + assert parent("hi") == "hi" + child = parent.fork() + agent.waiting = lambda: ["a pending parent answer"] + + assert child("research") == "research" + + # The stand-in echoes the complete prompt, so the return value proves no parent waiting + # prompt was appended to the child turn. + + +def test_a_second_fork_inherits_the_first_childs_frozen_context(claude: _Fake) -> None: + from dataclasses import replace + + agent = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="original", effort="low")) + parent = agent.new() + assert parent("hi") == "hi" + child = parent.fork() + agent.reconfigure(replace(agent.config, model="changed", effort="max")) + + grandchild = child.fork() + assert grandchild("research") == "research" + + grandchild_call = next( + call + for call in claude.calls() + if "--resume" in call["argv"] + and call["argv"][call["argv"].index("--resume") + 1] == grandchild.id + ) + assert ( + grandchild_call["argv"][grandchild_call["argv"].index("--model") + 1] + == "original" + ) + assert ( + grandchild_call["argv"][grandchild_call["argv"].index("--effort") + 1] == "low" + ) + + +# --- Codex: `thread/fork` on a dedicated server ---------------------------------------------- + + +def test_codex_forks_a_thread_and_the_child_runs_on_its_own_server( + codex: _Fake, +) -> None: + agent = CodexAgent(CodexAgentConfig(model="gpt-5-codex", effort="high")) + parent = agent.new() + assert parent("first") == "answered" + + child = parent.fork() + + assert child("research") == "answered" + assert parent("next") == "answered" # the parent goes on, unchanged + + calls = codex.calls() + forked = next(call for call in calls if call.get("method") == "thread/fork") + assert forked["params"]["threadId"] == parent.id + assert ( + "lastTurnId" not in forked["params"] + ) # forking through the latest completed turn + assert child.id == "child_thread" + assert parent.id == "parent_thread" + # Two servers: the parent's, and the child's dedicated one it overlaps on. + assert [call.get("method") for call in calls].count("initialize") == 2 + # The child's turn went to the child thread, not the parent's. + assert any( + call.get("method") == "turn/start" + and call["params"]["threadId"] == "child_thread" + for call in calls + ) + + +def test_codex_forks_through_an_earlier_completed_turn(codex: _Fake) -> None: + """An inclusive boundary: naming a completed turn forks through exactly that one.""" + agent = CodexAgent(CodexAgentConfig(model="gpt-5-codex", effort="high")) + parent = agent.new() + assert parent("first") == "answered" + assert parent("second") == "answered" + + assert parent.last_turn_id == "turn_fake" + child = parent.fork(last_turn_id=parent.last_turn_id) + + forked = next(call for call in codex.calls() if call.get("method") == "thread/fork") + assert forked["params"]["lastTurnId"] == "turn_fake" + assert child.id == "child_thread" + + +def test_reconfiguring_the_parent_does_not_change_a_codex_child(codex: _Fake) -> None: + """The first child turn carries the frozen values, not whatever the parent became.""" + from dataclasses import replace + + agent = CodexAgent(CodexAgentConfig(model="gpt-5-codex", effort="high")) + parent = agent.new() + assert parent("first") == "answered" + child = parent.fork() + + agent.reconfigure(replace(agent.config, model="changed", effort="max")) + assert child("research") == "answered" + + child_turn = next( + call + for call in codex.calls() + if call.get("method") == "turn/start" + and call["params"]["threadId"] == "child_thread" + ) + assert child_turn["params"]["model"] == "gpt-5-codex" + assert child_turn["params"]["effort"] == "high" + + +def test_a_codex_permission_override_is_sent_to_the_native_fork(codex: _Fake) -> None: + agent = CodexAgent(CodexAgentConfig(model="gpt-5-codex", effort="high")) + parent = agent.new() + assert parent("first") == "answered" + + child = parent.fork(permission="read-only") + + forked = next(call for call in codex.calls() if call.get("method") == "thread/fork") + assert forked["params"]["sandbox"] == "read-only" + assert forked["params"]["approvalPolicy"] == "never" + assert child._fork_context is not None + assert child._fork_context.cache_equivalent is False + + +def test_a_codex_fork_rejects_an_unknown_completed_boundary(codex: _Fake) -> None: + agent = CodexAgent(CodexAgentConfig(model="gpt-5-codex", effort="high")) + parent = agent.new() + assert parent("first") == "answered" + + with pytest.raises(RuntimeError, match="not a completed turn"): + parent.fork(last_turn_id="missing-turn") + + assert not [call for call in codex.calls() if call.get("method") == "thread/fork"] + + +def test_a_fork_child_error_is_not_hidden_by_suppress(claude: _Fake) -> None: + parent = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")).new() + assert parent("hi") == "hi" + child = parent.fork() + + with pytest.raises(subprocess.CalledProcessError): + child("boom", suppress=True) + + +def test_a_fork_child_does_not_enter_the_parent_fallback_chain( + claude: _Fake, monkeypatch: pytest.MonkeyPatch +) -> None: + from hmz.agents import Failed + + agent = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")) + parent = agent.new() + assert parent("hi") == "hi" + child = parent.fork() + + def fail(_prompt: str, *, schema: object = None) -> Any: + del schema + raise Failed(1, ["claude"], "", "child failed") + + monkeypatch.setattr(child, "_stream", fail) + monkeypatch.setattr( + agent, + "stands_in", + lambda: (_ for _ in ()).throw(AssertionError("fork child used fallback")), + ) + + with pytest.raises(Failed): + child("research") + + +def test_a_failed_fork_child_turn_is_recorded_without_transcript_content( + claude: _Fake, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from hmz.cycle import Cycle + + monkeypatch.chdir(tmp_path) + agent = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")) + cycle = Cycle("forking", [agent], "task", tmp_path) + agent.cycle = cycle + parent = agent.new() + assert parent("hi") == "hi" + child = parent.fork() + + with pytest.raises(subprocess.CalledProcessError): + child("boom", suppress=True) + + failures = [ + event for event in _events(cycle) if event.get("event") == "fork-failed" + ] + assert len(failures) == 1 + assert failures[0]["session_id"] == child.id + assert "boom" not in str(failures[0]) + + +# --- The run writes the fork down as a branch ------------------------------------------------- + + +def test_a_fork_is_written_down_as_a_branch_not_an_open( + claude: _Fake, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from hmz.cycle import Cycle, called, forks + + monkeypatch.chdir(tmp_path) + agent = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")) + cycle = Cycle("forking", [agent], "task", tmp_path) + agent.cycle = cycle + session = agent.new() + assert session("hi") == "hi" + + child = session.fork() + + forked = [event for event in _events(cycle) if event.get("event") == "forked"] + assert len(forked) == 1 + (said,) = forked + assert said["parent_session_id"] == session.id + assert said["session_id"] == child.id + assert said["parent_key"] == called(agent.id, "claude", "", session.id) + assert said["session_key"] == called(agent.id, "claude", "", child.id) + # And a trace can read the branch back: the child id maps to its parent id. + assert forks(cycle.path) == {child.id: session.id} + + +def test_permission_override_and_cache_equivalence_are_recorded( + claude: _Fake, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from hmz.cycle import Cycle + + monkeypatch.chdir(tmp_path) + agent = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")) + cycle = Cycle("forking", [agent], "task", tmp_path) + agent.cycle = cycle + parent = agent.new() + assert parent("hi") == "hi" + + child = parent.fork(permission="read-only") + + forked = [event for event in _events(cycle) if event.get("event") == "forked"] + assert forked[0]["session_id"] == child.id + assert forked[0]["permission"] == "read-only" + assert forked[0]["cache_equivalent"] is False + + +def _events(cycle: Any) -> list[dict[str, Any]]: + """Every line one cycle's own record wrote, in the order it wrote them.""" + from hmz.cycle import JOURNAL + + at = cycle.path / JOURNAL + return [json.loads(line) for line in at.read_text().splitlines()] + + +# --- The failure and retry probe: a response lost after the child was made ------------------- + + +#: A `claude` whose fork makes the child and exits without ever naming it back, which is how a +#: response lost after the backend created the branch is spelled. Every other turn behaves as +#: `_CLAUDE` does, so the parent can be opened and driven on. +_CLAUDE_LOST = """ +import json, pathlib, sys + +LOG = pathlib.Path(sys.argv[0] + ".log") + + +def note(entry): + with LOG.open("a") as stream: + json.dump(entry, stream) + stream.write("\\n") + + +def out(said): + print(json.dumps(said), flush=True) + + +argv = sys.argv[1:] +note({"argv": argv}) +flags = {} +for i, one in enumerate(argv): + if one.startswith("--") and i + 1 < len(argv) and not argv[i + 1].startswith("--"): + flags[one] = argv[i + 1] + elif one.startswith("--"): + flags[one] = True +if flags.get("--fork-session"): + sys.exit(0) # the child is made, but its name never comes back +sid = flags.get("--session-id") or flags.get("--resume") +out({"type": "system", "session_id": sid}) +for line in sys.stdin: + said = json.loads(line) + if said.get("type") != "user": + continue + text = said["message"]["content"][0]["text"] + out({"type": "assistant", "message": {"id": "msg_1", "role": "assistant", + "content": [{"type": "text", "text": text}]}}) + out({"type": "result", "subtype": "success", "is_error": False, "result": text}) +""" + +#: A `codex app-server` whose `thread/fork` fails after the branch was made, so the child id is +#: lost with the response. It is asked once, and the driver must not ask again. +_CODEX_FORK_LOST = """ +import json, pathlib, sys + +LOG = pathlib.Path(sys.argv[0] + ".log") + + +def send(message): + print(json.dumps(message), flush=True) + + +for line in sys.stdin: + call = json.loads(line) + with LOG.open("a") as stream: + json.dump(call, stream) + stream.write("\\n") + if "id" not in call: + continue + method = call.get("method") + result = {} + if method == "thread/start": + result = {"thread": {"id": "parent_thread"}} + elif method == "thread/fork": + send({"jsonrpc": "2.0", "id": call["id"], + "error": {"code": -32000, "message": "the child was made, then the stream broke"}}) + continue + send({"jsonrpc": "2.0", "id": call["id"], "result": result}) + if method == "thread/start": + send({"method": "thread/status/changed", + "params": {"status": {"type": "idle"}}}) + if method == "turn/start": + tid = call["params"].get("threadId", "parent_thread") + send({"method": "turn/started", + "params": {"turnId": "turn_fake", "threadId": tid}}) + send({"method": "item/completed", + "params": {"item": {"type": "agentMessage", "text": "answered"}, + "threadId": tid}}) + send({"method": "turn/completed", + "params": {"threadId": tid, + "turn": {"id": "turn_fake", "status": "completed"}}}) + send({"method": "thread/status/changed", + "params": {"status": {"type": "idle"}, "threadId": tid}}) +""" + + +def test_a_claude_fork_whose_response_is_lost_is_reconciled( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The child id is chosen up front, so a response lost with the process reconciles to it.""" + _install("claude", _CLAUDE_LOST, tmp_path, monkeypatch) + agent = ClaudeCodeAgent(ClaudeCodeAgentConfig(model="m", effort="low")) + parent = agent.new() + assert parent("hi") == "hi" + + child = parent.fork() # does not raise, though the fork never named itself + + assert child.id != parent.id # the id chosen up front, adopted regardless + + +def test_a_codex_fork_whose_response_is_lost_is_recorded_as_an_orphan( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The fork is asked once and never retried; the orphan is written down to reconcile.""" + from hmz.cycle import Cycle + + monkeypatch.chdir(tmp_path) + _install("codex", _CODEX_FORK_LOST, tmp_path, monkeypatch) + agent = CodexAgent(CodexAgentConfig(model="gpt-5-codex", effort="high")) + cycle = Cycle("forking", [agent], "task", tmp_path) + agent.cycle = cycle + parent = agent.new() + assert parent("first") == "answered" + + with pytest.raises(subprocess.CalledProcessError): + parent.fork() + + # Asked once, so a blind retry could not have made a second branch. + asked = [ + call + for call in _install_call_log(tmp_path, "codex") + if call.get("method") == "thread/fork" + ] + assert len(asked) == 1 + lost = [event for event in _events(cycle) if event.get("event") == "fork-lost"] + assert len(lost) == 1 + assert lost[0]["parent_session_id"] == parent.id + + +def _install_call_log(tmp_path: Path, name: str) -> list[dict[str, Any]]: + """Every call the stand-in `name` was made of.""" + return [ + json.loads(line) + for line in (tmp_path / "bin" / f"{name}.log").read_text().splitlines() + ] diff --git a/tests/tracing/test_collect.py b/tests/tracing/test_collect.py index e55f2974..c7cced4e 100644 --- a/tests/tracing/test_collect.py +++ b/tests/tracing/test_collect.py @@ -351,3 +351,15 @@ def test_keeps_unicode_readable( tracing.collect(workspace, output=tmp_path / "trace.json") assert "接上循环" in (tmp_path / "trace.json").read_text(encoding="utf-8") + + +def test_a_forked_child_is_drawn_under_its_parent( + claude_home: pathlib.Path, workspace: pathlib.Path +) -> None: + """A fork is a native branch, not a sub-agent, so the cycle's relation supplies the link.""" + document = collector.collect(workspace, parents={CLAUDE_SESSION: "parent-xyz"}) + + assert any( + event["args"].get("parent") == "claude:parent-xyz" + for event in banners(document) + ) diff --git a/tests/tracing/test_collect_command.py b/tests/tracing/test_collect_command.py index 1a8f7ea5..6cc12676 100644 --- a/tests/tracing/test_collect_command.py +++ b/tests/tracing/test_collect_command.py @@ -57,6 +57,7 @@ def run(*argv: str) -> int: "start": None, "end": None, "profile": None, # and nothing was profiled, there being no run + "parents": None, # nor forked, there being no run }, ), ( @@ -81,6 +82,7 @@ def run(*argv: str) -> int: "start": "1am", "end": "2am", "profile": None, + "parents": None, }, ), ],