From 246dd1beefc6b57a1e102b2a00395fd6293c7986 Mon Sep 17 00:00:00 2001 From: DBarr3 <143002219+DBarr3@users.noreply.github.com> Date: Wed, 12 Aug 2026 11:22:33 -0400 Subject: [PATCH] feat(cloud): bidirectional CloudBrain over the agent dev-session protocol The hosted path becomes a real coding brain: CloudBrain now creates an /agent/dev/sessions session, maps seq-numbered SSE frames (tool_call included) onto the existing BrainEvent vocabulary, POSTs locally executed tool results upstream (serialized, retried once against an idempotent server), and implements control(pause|resume|steer). close() aborts the stream and deletes the server session. Effort now rides the wire. Replay safety: frames at or below the last-seen seq are skipped, so a redelivered mutating tool_call can never execute twice; reconnect resumes from ?last_seq. Older servers (404/403 on create) fall back to the pre-existing one-way chat stream unchanged. Ships Phase A/B of the API coding IDE hardening spec: the current-state audit (AETHER_AGENT_API_CURRENT_STATE.md), dev-session route constants + GET SSE support in transport, the new frame vocabulary in stream.ts, and 10 protocol tests covering the round-trip, dedupe, reconnect, control, teardown, and both fallback splits. Co-Authored-By: Claude Fable 5 --- AETHER_AGENT_API_CURRENT_STATE.md | 163 ++++++++++++++ RELEASE_NOTES.md | 35 ++- src/core/brain_cloud.ts | 263 +++++++++++++++++++--- src/core/envelope.ts | 40 ++++ src/core/stream.ts | 52 ++++- src/core/transport.ts | 34 ++- test/brain_cloud.test.ts | 15 +- test/brain_cloud_dev.test.ts | 347 ++++++++++++++++++++++++++++++ 8 files changed, 908 insertions(+), 41 deletions(-) create mode 100644 AETHER_AGENT_API_CURRENT_STATE.md create mode 100644 test/brain_cloud_dev.test.ts diff --git a/AETHER_AGENT_API_CURRENT_STATE.md b/AETHER_AGENT_API_CURRENT_STATE.md new file mode 100644 index 0000000..a00d0a2 --- /dev/null +++ b/AETHER_AGENT_API_CURRENT_STATE.md @@ -0,0 +1,163 @@ +# AETHER AGENT — API CURRENT STATE (Phase A recon) + +**Date:** 2026-08-12 +**Agent baseline:** `AetherAI3/aether-agent` @ `eae6e28cdb6bdb9a1f8ab2d677285f5ce759f568` (main) +**Backend baseline:** `AetherAI3/AETHER-CLOUD` @ `e96ad64c` (origin/main) +**Purpose:** Phase A deliverable for the API Coding IDE Hardening Spec (§121). Maps what exists today, confirms the parity gap, and records the constraints Phase B–L must respect. + +--- + +## 1. Agent Brain protocol + +- `Brain` interface (`src/core/brain.ts:17-26`): `run(task)` → `AsyncIterable`, `sendToolResult(id, result)`, `control("pause"|"resume"|"steer", note?)`, `close()`. +- `BrainEvent` union (`src/core/brain_protocol.ts:66-118`): `stage, monologue, skill, turn, tool_call, telemetry, status, checkpoint, done, error, memory` + 7 workflow-swarm frames. +- `HostCommand` (`brain_protocol.ts:121-124`): `task | tool_result | control`. `PROTOCOL_VERSION = 3`, pinned by `test/fixtures/bridge_conformance.json`. +- Three implementations: + +| impl | transport | `sendToolResult` | `control` | +|---|---|---|---| +| `LocalBrain` (`brain_local.ts`) | Python child, NDJSON stdio | real | real | +| `OllamaBrain` (`brain_ollama.ts`) | in-process TS loop over Ollama HTTP | real (but ignores tool-call id — FIFO assumption, `:80-86`) | no-op | +| `CloudBrain` (`brain_cloud.ts`) | one-way SSE | **no-op (`:90`)** | **no-op (`:91`)** | + +## 2. CloudBrain — the confirmed parity gap + +- POSTs `ChatWireRequest {query, forced_model_key, agent_name, model_pick_source}` (`envelope.ts:41`) to `/agent/chat/stream`; fail-soft non-stream fallback to `/agent/chat` on `StreamUnavailableError`. +- **Never maps a `tool_call` frame** — the host's execute/gate branch is dead code on the cloud path. Server runs tools server-side (or not at all). +- `close()` sets a flag only; never aborts the fetch (`api.stream()` accepts an AbortSignal CloudBrain doesn't pass). +- Synthesizes its own terminal `done`; never trusts server `done`. +- `custody` frames persisted via `appendCustody()`; `usage/ping/progress/...` dropped. +- **`TaskCommand.effort` is never forwarded to the cloud** — `buildChatRequest` has no effort field, despite `/effort` docs claiming it rides to the cloud brain. + +## 3. Host loop & verification + +- `cmdCode` (`src/commands/code.ts:102-395`): workspace resolution (`--repo`/`--worktree` vs `prepareWorkspace` gate) → backend choice → `LocalBrain` or `CloudBrain` (`OllamaBrain` unreachable from `aether agent`) → `ToolExecutor` → `hostLoop` (`:403-445`). +- Permission gate: `decideGate` (`src/core/autonomy.ts:70-82`) maps tool → action via `TOOL_DEFINITIONS[name].sideEffect`; only `write|shell|git` gate. **`network` side-effect (web_search/web_fetch) always allowed, never prompted.** Fail-closed on non-TTY without `--yes`. +- Host verification: `finalVerify` (`src/core/verify_gate.ts:60-84`) re-runs the configured test command and derives status from the real exit code. `errored` overrides green; no testCmd → `unverified`. Model "done" is advisory only. +- No host-side tool-call budget, per-tool timeout, or re-entrancy guard in `hostLoop`. + +## 4. Local tools (ToolExecutor, `src/core/tool_executor.ts`) + +- 8 canonical tools (`brain_protocol.ts:127-137`): `read_file, write_file, run_shell, run_tests, repo_search, git_commit, web_search, web_fetch`. +- Typed arg validation before any side effect (`tool_registry.ts:82-118`): unknown tool/extra args/oversize rejected; coverage asserted by test. +- Path confinement `safe()` (realpath nearest-existing-ancestor vs root); `writeFile` uses `O_NOFOLLOW` + pre-open revalidation; `repoSearch` skips symlinks; **`readFile` lacks `O_NOFOLLOW`**. +- Caps: output 8000 chars (head/tail for tests), snapshot 1 MiB, search 40 hits; `run()` = `spawnSync shell:true`, 15-min timeout, 64 MiB buffer. +- `git_commit` via `GitCommitGuard` — argv only, refuses pre-staged/unexpected staged/>1000 paths. +- Pre-write snapshots feed live diff rendering (rendered **before** gate decision — denied writes still paint a diff). + +## 5. Transport / auth + +- `ApiClient` (`src/core/transport.ts`, 717 lines). Base `https://api.aethersystems.net/cloud` (`AETHER_BASE_URL` override, never persisted). +- Token: `FileTokenStore` `configDir()/.token`, `O_NOFOLLOW`, 0600; env/static stores; `aek_` API keys. +- Bearer refused over cleartext except localhost; cross-origin `getBinary` goes unauthenticated; server text sanitized (C0/C1 strip, 200-char cap). +- SSE: POST + `Accept: text/event-stream`; JSON response → `StreamUnavailableError` (fail-soft contract); idle-interval timeout (120 s stream / 30 s request, env-overridable, 0=off); unknown frame types → `null` by contract. +- Retry: refresh-on-401 once, shared in-flight; **no retry for 5xx/network/timeout anywhere; no reconnect/resume concept.** + +## 6. GitHub — two disjoint paths + +- (a) Backend App link (`src/core/github.ts`, `commands/github.ts`): `status|connect|disconnect` against `/account/github/{status,connect,disconnect}` + poll. No local GitHub token ever. +- (b) `--repo owner/name` (`src/core/repo.ts`): user's own `gh`/`git` auth; mirror `~/.aether-agent/repos/-`; **reused mirror never fetched — arbitrarily stale**; PR = printed `gh pr create` hint. +- No PR/checks/reviews/security context anywhere in the CLI. + +## 7. Repo / worktree + +- Two divergent flows (`src/core/worktree.ts`): flag-driven (`~/.aether-agent/worktrees`, branch `aether/-`) vs gate-driven (`configDir()/worktrees`, `aether/[-N]`, collision retry ×5, degrade to run-in-place). No listing/pruning; no cleanup. + +## 8. Session / context + +- `ContextRegistry` (`src/core/context_registry.ts`) — module singleton: pins/drops/uvtCap/plan/label/HUD. **UVT cap display-only: `uvtSpent` never incremented, `checkUvtCap()` zero callers.** +- Snapshots workspace-scoped + traversal-proof (`workspace_scope.ts`); backend sync `POST/GET /agent/context` fail-soft. +- `SessionLog` `~/.aether-agent/logs/...` with substantive redaction (bearer/token/secret patterns, content/command omitted); resume replays events, workspace-scoped. +- All of this is **session context**, not live engineering context (no Git snapshot, PR, CI, security state). + +## 9. Model / effort + +- No local model list; catalog from `GET /models`. Effort tiers `LOW..CODEPRO` (`src/ui/effort.ts:16`) mirror `lib/orchestrator/presets/contracts.py`. Effort reaches the local Python brain only (see §2). + +## 10. AETHER-CLOUD — agent/chat route + +- `api/routes/agent_core_routes.py:621` — `POST /agent/chat/stream`, rate-limited, Protocol-C-gated, dual token resolve. **Strictly one-way StreamingResponse**; no tool-result ingress on this route. +- Existing tool-result ingress lives elsewhere: `POST /agent/coding/step` (desktop; client sends whole message array, server returns `tool_uses`), `POST /agent/run/{id}/subtask/{sid}/result` (orchestrator), `POST /agent/code/permission/{id}/decide` (web Code). +- Pre-stream pipeline: prompt-injection guard → UVT preflight (`pricing_guard.preflight`, Gates 0–4) → forced-model tier validation → Protocol-C custody commit → producer → WorkflowGate swarm branch (HIGH/CODEPRO). + +## 11. AETHER-CLOUD — SSE protocol + +- `lib/sse_protocol.py` = SSOT: `data:{json}\n\n` only, no `event:` names. Vocabulary: `open, ping, reasoning, delta (incremental-only contract), usage, done, error, turn_outcome, workspace_edit_*, user_uvt_remaining, custody, command_result, media, session_status, notice, research_*`. +- **Three incompatible framings in the wild:** chat (`data:`-only), orchestrator route (`event:`+`data:`, `: ping` comments), web Code loop (hand-rolled dicts, its `tool_call/permission_request/tool_result/tool_denied` vocabulary absent from the SSOT module). + +## 12. AETHER-CLOUD — web Code loop (do NOT copy into terminal) + +- `code_routes.py` (repo root, outside `api/routes/` pattern; own auth helpers, own SSE serializer, tier gate `return True`). +- `lib/code/agent_loop.py:run_code_agent` — Anthropic tool-use loop, server worktree under `/conversations//worktree/`, `_MAX_STEPS=24`, permission Future registry (`lib/code/permission_registry.py`, **process-local — breaks under >1 worker**), 180 s permission timeout, tool output re-scanned by prompt guard before re-entering model. +- Tool schema: `lib/orchestrator/core/coding_tools.py` — `CODING_TOOL_SCHEMAS` (~28 tools, shared with desktop), `READ_ONLY_TOOLS`, `CODE_AGENT_TOOLS` (web subset; excludes ssh/scp/git_push etc. — server identity), `is_mutating()` fail-closed. **This is the schema to normalize against (spec §15), not the loop to reuse (spec §14).** + +## 13. Model routing / CODEPRO / billing + +- Selection `lib/router.py` (typed tripwires, not policy); authoritative policy = TS PolicyGate `/api/internal/router/pick`, fails closed. +- All provider HTTP in `lib/token_accountant.py`; retry = transport errors only, never streams; usage idempotent on `request_id` (`ON CONFLICT DO NOTHING` — the idempotency primitive Phase B can reuse). +- UVT gates: `lib/pricing_guard.py` `preflight()` Gates 0–4 + slot lifecycle. Live `usage`/`user_uvt_remaining` frames during stream, authoritative DB read after `done`. +- CODEPRO: `resolve_effort_mode()` (`lib/deep_thinking.py`), session caps `lib/orchestrator/atlas/codepro_caps.py`, swarm via `WorkflowGate`. +- CI/Actions billing: `lib/ci_billing/` quote/authorize/record/settle/release; `predator-replay-v1` rate card. + +## 14. GitHub App (backend) + +- `lib/github_app.py`: JWT, installation token mint with cache + invalidation, repo-scope narrowing; `lib/github_connect_routes.py`: connect/manage/status/repositories/callback/disconnect/webhook. +- Actions adapter `lib/actions_hosted/github.py`: per-repo token narrowed to exact `{contents:read, checks:write}`, fail-closed on grant mismatch, never persisted/logged. +- **Gap: no Dependabot or code-scanning read helpers exist anywhere.** Repo reads limited to installation repo list + Actions tarball/check/PR calls. Phase E must build these. + +## 15. Aether Actions / Predator + +- `api/routes/actions_hosted_routes.py`: owner plane (quote/runs/settings/PRs/artifacts, run nonce + `approvedMaximumUvt`) + worker plane (lease claim/renew, internal source/events/artifacts/complete). Executor registry pinned by image digest. +- Predator = `predator.replay` operation kind inside Actions (not separate endpoints). **No server-side findings or certificate model exists** — results are artifacts + a "Predator Security" GitHub Check. Phase K depends on this being built. + +## 16. Project context / memory / custody + +- Project identity `lib/orchestrator/memory/project_identity.py` (uuid5, flag `AETHER_PROJECT_BIND_ENABLED`); read path fail-soft, sanitized, clipped. +- QOPC routes/memory; web-Code typed memory nodes (`lib/code/memory.py`). +- Protocol-C: ASGI stamp middleware (SSE-safe) + per-turn COMMITMENT/ATTESTATION custody frames (client-held, never stored server-side) + orchestrator audit sink. `_require_protocol_c` 503s model routes when signer unhealthy. + +## 17. Router extraction pattern (Phase B must follow) + +1. New module `api/routes/_routes.py`, bare `APIRouter()` or cohesive prefix. +2. Leaf deps at top (`api/deps_protocol.py`, `api/deps_session.py`, `lib/_rate_limit.py`); `api_server` symbols imported lazily inside handlers. +3. Mount at bottom of `api_server.py`; registration order matters. +4. `tests/api/test_openapi_snapshot.py` guards the contract — new routes update the snapshot deliberately, refactors must not. +5. Anti-pattern on record: root-level `code_routes.py`/`nano_routes.py`. + +## 18. Test infrastructure + +- Agent: `node:test`, zero runtime deps, tests compiled by tsc and run from `dist/` (`npm test`), 110 files; bridge conformance fixture pins protocol v3; `npm run smoke` = 7-check harness; CI ubuntu+windows Node 24, SHA-pinned actions, SBOM. +- Backend: pytest (asyncio auto), 896 files; tier0–3 markers; route tests inject fake `api_server` into `sys.modules` + in-memory stores; OpenAPI snapshot guard; `tests/parity/` runs desktop/backend tool-schema parity. + +--- + +## 19. Overlap with open work + +- aether-agent open PRs at baseline: #62/#63/#64 (Dependabot GH-Actions bumps), #36 (old docs spec). **None own this lane** — spec §121 assumption re-verified 2026-08-12. +- AETHER-CLOUD local checkout was ~3.6k files behind origin/main; all lane-relevant surfaces exist only on origin/main. Sync before backend work. + +## 20. Defect register feeding Phases B–H + +Agent side: +1. `CloudBrain.sendToolResult`/`control` no-ops; no `tool_call` mapping (Phase B — the gap). +2. `close()` doesn't abort the stream (Phase B). +3. Effort never sent to cloud (Phase B request contract). +4. No reconnect/resume, no 5xx/network retry (Phase B §80). +5. `web_search`/`web_fetch` never gated (Phase H permission categories, spec §59). +6. UVT cap display-only (Phase I, spec §49). +7. `hostLoop` unbounded: no tool budget/per-tool timeout (Phase C/G). +8. `readFile` no `O_NOFOLLOW` (Phase C). +9. Gate prompt truncates command at 200 chars — user approves partially visible command (Phase H, spec §61). +10. Diff rendered before gate decision (Phase H). +11. `--repo` mirror never refreshed; two worktree roots, no prune (Phase E/D). +12. `OllamaBrain.sendToolResult` ignores id — blocks any parallel tool dispatch (Phase C). +13. Ollama tool schema (`buildToolSchemas`) drifts from `TOOL_DEFINITIONS` (Phase C parity). +14. `parseArgs strict:false` swallows typo'd flags (Phase H). + +Backend side: +1. No bidirectional dev-session protocol — build `/agent/dev/sessions` family per spec §7 (Phase B). +2. SSE framing not unified; Code vocabulary outside SSOT (Phase B event vocabulary, spec §10). +3. Permission registry process-local/unbounded (design constraint for the new session store). +4. No Dependabot/code-scanning readers (Phase E). +5. No Predator findings/certificate model (Phase K). +6. Code-loop UVT preflight non-denying (do not inherit into dev sessions). diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 0f65f79..667131b 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -1,4 +1,37 @@ -# Aether Agent — August 2026 fleet update +# Aether Agent — the API brain goes bidirectional + +**August 12, 2026** + +The hosted path is now a true coding brain for your terminal. Until today the +cloud stream was one-way: the server reasoned and replied, but it could not +drive the tools on your machine. That gap is closed. + +- **Agent dev sessions** — `aether agent` on the API path now opens a dedicated + coding session: the Aether API plans and reasons, and every file read, edit, + shell command, test run, and Git commit executes locally through the same + permission gate and path guards the local brain has always used. Your source + tree never leaves your machine. +- **Replay-safe by construction** — every frame carries a per-session sequence + number. After a network drop the Agent reconnects from the last frame it saw, + and a redelivered `tool_call` is skipped, never re-executed. Tool results are + idempotent upstream: a retried POST is a no-op, a conflicting one is refused. +- **Steer, pause, resume — for real** — `/steer`, pause, and resume now reach + the hosted brain mid-session and apply at the next model step. No restart, no + lost work. +- **Effort reaches the cloud** — the `/effort` dial (LOW through CODEPRO) is now + on the wire for hosted runs; previously it only shaped local runs despite the + docs saying otherwise. +- **Clean teardown** — Ctrl+C/exit now aborts the stream immediately and closes + the server session instead of leaving the socket to idle out for two minutes. +- **Graceful fallback** — against an older server (or with the feature flag + off) the Agent silently uses the previous one-way stream. Nothing breaks. + +Requires a server with agent dev sessions enabled; the client negotiates the +protocol version at session start and fails safe. + +--- + +## Aether Agent — August 2026 fleet update **August 6, 2026** diff --git a/src/core/brain_cloud.ts b/src/core/brain_cloud.ts index 342fac1..9ae337f 100644 --- a/src/core/brain_cloud.ts +++ b/src/core/brain_cloud.ts @@ -1,27 +1,62 @@ -// CloudBrain — the cloud transport. Surfaces the existing AetherCloud universal -// SSE stream through the SAME BrainEvent vocabulary the local brain emits, so -// the host renders cloud and local runs identically. +// CloudBrain — the cloud transport. Surfaces the Aether API through the SAME +// BrainEvent vocabulary the local brain emits, so the host renders cloud and +// local runs identically. // -// Honest boundary (today's server contract): the universal stream is one-way -// and runs its tools server-side, so it does NOT yet emit `tool_call` frames or -// accept an upstream `tool_result`. CloudBrain therefore maps the frames that -// exist (delta/reasoning/task_*/done/error) and `sendToolResult` is a no-op. -// When the server adds tool_call frames + an upstream channel, this class -// implements the same round-trip the local brain already does — no host change. +// Primary path (dev-session protocol): POST /agent/dev/sessions creates a +// server-side coding session; GET .../stream delivers seq-numbered frames +// (tool_call included); the host executes locally and sendToolResult() POSTs +// the result back (idempotent server-side). control() maps pause/resume/steer +// onto POST .../control. This is the real bidirectional round-trip: the API is +// the brain, this machine is the authority — code stays local. +// +// Fallback path (legacy one-way chat stream): when the server has no dev +// route (404) or the feature is off (403), CloudBrain degrades to the +// pre-existing /agent/chat/stream behavior, where sendToolResult/control are +// documented no-ops because the server runs its tools server-side. import type { Brain, TaskCommand } from "./brain.js"; import { EventQueue } from "./brain.js"; import type { BrainEvent } from "./brain_protocol.js"; +import { TOOLS } from "./brain_protocol.js"; +import type { ToolResult } from "./tool_executor.js"; import type { ApiClient } from "./transport.js"; -import { CHAT_STREAM_PATH, CHAT_PATH, defaultStreamTimeoutMs } from "./transport.js"; -import { buildChatRequest } from "./envelope.js"; +import { + CHAT_STREAM_PATH, + CHAT_PATH, + DEV_SESSIONS_PATH, + devSessionStreamPath, + devSessionToolResultsPath, + devSessionControlPath, + devSessionPath, + defaultStreamTimeoutMs, +} from "./transport.js"; +import { buildChatRequest, buildDevSessionRequest } from "./envelope.js"; import { decodeSse, type StreamFrame } from "./stream.js"; -import { StreamIncompleteError, StreamUnavailableError } from "./errors.js"; +import { HttpError, StreamIncompleteError, StreamUnavailableError } from "./errors.js"; import { appendCustody } from "./custody.js"; import { hintFor } from "./error_hints.js"; +/** Version of the dev-session wire protocol this client speaks. */ +export const DEV_PROTOCOL_VERSION = 1; + +/** Reconnect budget for a dropped dev-session stream (resets on progress). */ +const MAX_RECONNECTS = 3; +const RECONNECT_DELAY_MS = 1_000; + +interface DevSessionCreated { + session_id: string; + protocol_version: number; + model?: string; + tools?: string[]; +} + export class CloudBrain implements Brain { private aborted = false; + private net: AbortController | null = null; + private sessionId: string | null = null; + private lastSeq = 0; + /** Serializes upstream result POSTs so they arrive in execution order. */ + private upstream: Promise = Promise.resolve(); constructor(private readonly api: ApiClient) {} @@ -32,20 +67,119 @@ export class CloudBrain implements Brain { } private async pump(task: TaskCommand, queue: EventQueue): Promise { + try { + let created: DevSessionCreated | null = null; + try { + created = await this.api.postJson( + DEV_SESSIONS_PATH, + buildDevSessionRequest({ + task: task.text, + model: task.model, + effort: task.effort, + capabilities: TOOLS, + protocolVersion: DEV_PROTOCOL_VERSION, + }), + ); + } catch (err) { + if (isLegacyServer(err)) { + await this.legacyPump(task, queue); + return; + } + throw err; + } + this.sessionId = created.session_id; + queue.push({ type: "stage", name: "execute", face: "⟨◉⟩" }); // uplink face + await this.devPump(queue); + } catch (err) { + queue.push({ type: "error", msg: withHint(err) }); + } finally { + queue.end(); + } + } + + /** Consume the dev-session stream, reconnecting from lastSeq on a drop. */ + private async devPump(queue: EventQueue): Promise { + const id = this.sessionId as string; + let failed: string | null = null; + let sawDone = false; + let doneOk = true; + let reconnects = 0; + + while (!this.aborted && !sawDone && !failed) { + this.net = new AbortController(); + let progressed = false; + try { + const stream = await this.api.stream(devSessionStreamPath(id, this.lastSeq), undefined, { + method: "GET", + signal: this.net.signal, + }); + for await (const frame of decodeSse(stream)) { + if (this.aborted) break; + // Duplicate delivery is safe by contract: a replayed frame keeps its + // seq, so anything at or below the high-water mark is skipped and a + // mutating tool_call can never execute twice. + if (typeof frame.seq === "number") { + if (frame.seq <= this.lastSeq) continue; + this.lastSeq = frame.seq; + progressed = true; + reconnects = 0; + } + if (frame.type === "custody") appendCustody(frame.custody); + if (frame.type === "error") failed = frame.msg; + if (frame.type === "done") { + sawDone = true; + doneOk = frame.ok !== false; + } + const ev = mapDevFrame(frame); + if (ev) queue.push(ev); + if (sawDone || failed) break; + } + } catch (err) { + if (this.aborted) break; + if (reconnects >= MAX_RECONNECTS) { + failed = withHint(err); + break; + } + reconnects += 1; + await sleep(RECONNECT_DELAY_MS * reconnects); + continue; + } + if (this.aborted || sawDone || failed) break; + // Stream ended cleanly without a terminal frame — reconnect from seq + // unless the budget is spent (a quiet server keeps its socket open, so + // a clean early end is either a proxy hiccup or a dead session). + if (reconnects >= MAX_RECONNECTS && !progressed) { + failed = withHint(new StreamIncompleteError()); + break; + } + reconnects += 1; + await sleep(RECONNECT_DELAY_MS * reconnects); + } + + if (failed) { + queue.push({ type: "done", ok: false, result: failed, remaining: 0, reason: "" }); + } else if (this.aborted) { + queue.push({ type: "done", ok: false, result: "aborted", remaining: 0, reason: "" }); + } else { + queue.push({ type: "done", ok: doneOk, result: "", remaining: 0, reason: "" }); + } + } + + /** The pre-dev-protocol one-way chat stream (older/flagged-off servers). */ + private async legacyPump(task: TaskCommand, queue: EventQueue): Promise { const req = buildChatRequest({ prompt: task.text, model: task.model || undefined, manualModel: Boolean(task.model), }); - queue.push({ type: "stage", name: "execute", face: "⟨◉⟩" }); // uplink face + queue.push({ type: "stage", name: "execute", face: "⟨◉⟩" }); try { - const stream = await this.api.stream(CHAT_STREAM_PATH, req); + this.net = new AbortController(); + const stream = await this.api.stream(CHAT_STREAM_PATH, req, { signal: this.net.signal }); // The terminal done must be ground truth, never fabricated success - // (CONTRACTS.md invariant 5): a streamed error, a user abort, or the - // stream ending without ever sending a terminal done/error frame (a - // clean-looking premature close — LOOP-06 round 3) all end the run - // ok:false. Custody receipts persist here too — the server stores - // nothing; the client-held log is the only copy. + // (CONTRACTS.md invariant 5): a streamed error, a user abort, or a + // clean-looking premature close all end the run ok:false. Custody + // receipts persist here too — the client-held log is the only copy. let failed: string | null = null; let sawDone = false; for await (const frame of decodeSse(stream)) { @@ -53,7 +187,7 @@ export class CloudBrain implements Brain { if (frame.type === "custody") appendCustody(frame.custody); if (frame.type === "error") failed = frame.msg; if (frame.type === "done") sawDone = true; - const ev = mapFrame(frame); + const ev = mapLegacyFrame(frame); if (ev) queue.push(ev); } if (failed) { @@ -69,8 +203,7 @@ export class CloudBrain implements Brain { if (err instanceof StreamUnavailableError) { // Fail-soft: non-streaming fallback (contract `{"stream": false}`). A // full LLM turn can legitimately run long, so this opts into - // stream()'s own generous bound rather than request()'s 30s - // metadata-call default (LOOP-01/LOOP-06 round-1). + // stream()'s own generous bound rather than request()'s 30s default. try { const r = await this.api.postJson<{ response?: string }>(CHAT_PATH, req, undefined, defaultStreamTimeoutMs()); queue.push({ type: "monologue", text: r.response ?? "", depth: 0 }); @@ -81,29 +214,97 @@ export class CloudBrain implements Brain { } else { queue.push({ type: "error", msg: withHint(err) }); } - } finally { - queue.end(); } } - // The cloud stream executes tools server-side today; nothing to send back. - sendToolResult(): void {} - control(): void {} + /** POST a locally executed tool result upstream. Serialized so results land + * in execution order; retried once (the server accepts duplicates as + * idempotent no-ops, so a lost ack can never double-apply). */ + sendToolResult(id: string, result: ToolResult): void { + const sessionId = this.sessionId; + if (!sessionId) return; // legacy path: server ran the tool itself + const body = { + tool_call_id: id, + status: result.exitCode === 0 ? "ok" : "error", + exit_code: result.exitCode, + output: result.output, + truncated: false, + }; + this.upstream = this.upstream.then(async () => { + const path = devSessionToolResultsPath(sessionId); + try { + await this.api.postJson(path, body); + } catch { + if (this.aborted) return; + await sleep(RECONNECT_DELAY_MS); + try { + await this.api.postJson(path, body); + } catch { + // Deliberately swallowed: the server's tool-result timeout will + // terminate the session with a clear error frame; throwing here + // would only crash the host loop mid-render. + } + } + }); + } + + control(action: "pause" | "resume" | "steer", note?: string): void { + const sessionId = this.sessionId; + if (!sessionId) return; // legacy path: no server session to control + void this.api + .postJson(devSessionControlPath(sessionId), { action, note: note ?? null }) + .catch(() => {}); // fire-and-forget; a lost steer is re-typeable + } + close(): void { this.aborted = true; + this.net?.abort(); + const sessionId = this.sessionId; + if (sessionId) { + void this.api.deleteJson(devSessionPath(sessionId)).catch(() => {}); + } } } -/** Error message plus its recovery hint (if any), e.g. for a StreamTimeoutError: - * "stream timed out after 120s with no data (the stream went quiet - retry, or /doctor to check connectivity)". */ +/** Error message plus its recovery hint (if any). */ function withHint(err: unknown): string { const msg = err instanceof Error ? err.message : String(err); const hint = hintFor(err); return hint ? `${msg} (${hint})` : msg; } -/** Map a universal SSE frame onto the bridge event vocabulary (null = ignore). */ -function mapFrame(f: StreamFrame): BrainEvent | null { +/** An older/flagged-off server: no dev-session route. 404 = route absent, + * 403 = AETHER_AGENT_DEV_ENABLED off — both degrade to the legacy stream. */ +function isLegacyServer(err: unknown): boolean { + return err instanceof HttpError && (err.status === 404 || err.status === 403); +} + +function sleep(ms: number): Promise { + return new Promise((r) => setTimeout(r, ms)); +} + +/** Map a dev-session frame onto the bridge event vocabulary (null = ignore). */ +function mapDevFrame(f: StreamFrame): BrainEvent | null { + switch (f.type) { + case "session": + return { type: "status", phase: `session ${f.model ?? ""}`.trim(), poolUsed: 0, poolCap: 0 }; + case "tool_call": + return { type: "tool_call", id: f.toolCallId, name: f.name, args: f.args }; + case "reasoning": + return { type: "monologue", text: f.text, depth: 1 }; + case "delta": + return { type: "monologue", text: f.text, depth: 0 }; + case "error": + return { type: "error", msg: f.msg }; + case "done": + return null; // devPump emits its own terminal done + default: + return null; // open/ping/usage/tool_result_ack/custody — not agent view + } +} + +/** Map a legacy universal SSE frame onto the bridge vocabulary (null = ignore). */ +function mapLegacyFrame(f: StreamFrame): BrainEvent | null { switch (f.type) { case "reasoning": return { type: "monologue", text: f.text, depth: 1 }; diff --git a/src/core/envelope.ts b/src/core/envelope.ts index 16b19e7..01e22df 100644 --- a/src/core/envelope.ts +++ b/src/core/envelope.ts @@ -51,6 +51,46 @@ export function buildChatRequest(args: BuildChatRequestArgs): ChatWireRequest { return req; } +// ── Agent dev sessions (POST /agent/dev/sessions) ─────────────────────────── + +/** Wire shape bound by AETHER-CLOUD api/routes/agent_dev_session_routes.py. */ +export interface DevSessionWireRequest { + task: string; + surface: "aether_agent"; + model: string | null; + effort: string | null; + /** Tool names this host supports — the client owns the allowlist; the + * server intersects with its own known set and never sends anything else. */ + capabilities: string[]; + max_uvt?: number; + repo?: Record; + protocol_version: number; +} + +export interface BuildDevSessionArgs { + task: string; + model?: string; + effort?: string; + capabilities: readonly string[]; + maxUvt?: number; + repo?: Record; + protocolVersion: number; +} + +export function buildDevSessionRequest(args: BuildDevSessionArgs): DevSessionWireRequest { + const req: DevSessionWireRequest = { + task: args.task, + surface: "aether_agent", + model: args.model?.trim() || null, + effort: args.effort?.trim() || null, + capabilities: [...args.capabilities], + protocol_version: args.protocolVersion, + }; + if (args.maxUvt && args.maxUvt > 0) req.max_uvt = args.maxUvt; + if (args.repo) req.repo = args.repo; + return req; +} + // Local-only coding envelope (kept for the future coding route + workspace). export interface CodingEnvelope { prompt: string; diff --git a/src/core/stream.ts b/src/core/stream.ts index 29b5b40..cb73c2b 100644 --- a/src/core/stream.ts +++ b/src/core/stream.ts @@ -9,15 +9,28 @@ // be ignored. The CF-flush preamble (`:<4096 spaces>`) and `ping` heartbeat are // handled here (comment lines skipped; ping surfaced as a typed liveness frame). -export type StreamFrame = +export type StreamFrame = StreamFrameBody & { + /** Per-session monotonic sequence number (dev-session frames only). A + * reconnecting client resumes with ?last_seq=N and MUST skip seq <= N so a + * replayed mutating tool_call is never executed twice. */ + seq?: number; +}; + +export type StreamFrameBody = // shared | { type: "open" } | { type: "ping" } | { type: "reasoning"; text: string } | { type: "delta"; text: string } | { type: "usage"; uvt: number; cents: number } - | { type: "done"; uvt: number; cents: number; inputTokens?: number; outputTokens?: number } + | { type: "done"; uvt: number; cents: number; inputTokens?: number; outputTokens?: number; ok?: boolean } | { type: "error"; msg: string; errorCode?: string; refId?: string } + // Agent dev sessions (/agent/dev/sessions/{id}/stream) — the bidirectional + // coding protocol: the API brain emits tool_call, the local host executes + // and POSTs the result back. + | { type: "session"; sessionId: string; protocolVersion: number; model?: string; tools?: string[] } + | { type: "tool_call"; toolCallId: string; name: string; args: Record; risk?: string } + | { type: "tool_result_ack"; toolCallId: string } // The server-signed chain-of-custody for this turn (commitment + attestation). // The server signs but never stores it — the client decides whether to persist // (the CLI logs it locally; a web client may show-then-discard). @@ -77,6 +90,15 @@ export type StreamFrame = /** Normalize a parsed JSON object (snake_case wire → camelCase) into a frame. */ export function normalizeFrame(obj: Record): StreamFrame | null { + const frame = normalizeFrameBody(obj) as StreamFrame | null; + if (frame) { + const seq = obj["seq"]; + if (typeof seq === "number" && Number.isFinite(seq)) frame.seq = seq; + } + return frame; +} + +function normalizeFrameBody(obj: Record): StreamFrameBody | null { const type = obj["type"]; if (typeof type !== "string") return null; switch (type) { @@ -97,14 +119,38 @@ export function normalizeFrame(obj: Record): StreamFrame | null cents: Number(obj["cents"] ?? 0), inputTokens: numOrUndef(obj["input_tokens"] ?? obj["inputTokens"]), outputTokens: numOrUndef(obj["output_tokens"] ?? obj["outputTokens"]), + // Only set when the wire carried it (dev-session done frames) — legacy + // frames must round-trip byte-identical for the conformance tests. + ...(obj["ok"] === undefined ? {} : { ok: Boolean(obj["ok"]) }), }; case "error": return { type: "error", msg: String(obj["msg"] ?? obj["message"] ?? ""), - errorCode: strOrUndef(obj["error_code"] ?? obj["errorCode"]), + errorCode: strOrUndef(obj["error_code"] ?? obj["errorCode"] ?? obj["code"]), refId: strOrUndef(obj["ref_id"] ?? obj["refId"]), }; + case "session": + return { + type: "session", + sessionId: String(obj["session_id"] ?? obj["sessionId"] ?? ""), + protocolVersion: Number(obj["protocol_version"] ?? obj["protocolVersion"] ?? 0), + model: strOrUndef(obj["model"]), + tools: parseStrArray(obj["tools"]), + }; + case "tool_call": + return { + type: "tool_call", + toolCallId: String(obj["tool_call_id"] ?? obj["toolCallId"] ?? ""), + name: String(obj["name"] ?? ""), + args: (obj["args"] as Record) ?? {}, + risk: strOrUndef(obj["risk"]), + }; + case "tool_result_ack": + return { + type: "tool_result_ack", + toolCallId: String(obj["tool_call_id"] ?? obj["toolCallId"] ?? ""), + }; case "custody": return { type: "custody", diff --git a/src/core/transport.ts b/src/core/transport.ts index 26301cd..b410094 100644 --- a/src/core/transport.ts +++ b/src/core/transport.ts @@ -55,6 +55,22 @@ export function isSameOrigin(target: string, baseUrl: string): boolean { // Aether API routes. export const CHAT_STREAM_PATH = "/agent/chat/stream"; // standard chat SSE export const CHAT_PATH = "/agent/chat"; // non-streaming fail-soft fallback +// Agent dev sessions — the bidirectional coding protocol (API brain, local +// host): downstream SSE with per-session `seq`, upstream idempotent POSTs. +export const DEV_SESSIONS_PATH = "/agent/dev/sessions"; +export function devSessionStreamPath(id: string, lastSeq = 0): string { + const base = `${DEV_SESSIONS_PATH}/${encodeURIComponent(id)}/stream`; + return lastSeq > 0 ? `${base}?last_seq=${lastSeq}` : base; +} +export function devSessionToolResultsPath(id: string): string { + return `${DEV_SESSIONS_PATH}/${encodeURIComponent(id)}/tool-results`; +} +export function devSessionControlPath(id: string): string { + return `${DEV_SESSIONS_PATH}/${encodeURIComponent(id)}/control`; +} +export function devSessionPath(id: string): string { + return `${DEV_SESSIONS_PATH}/${encodeURIComponent(id)}`; +} // Auth (session_token via username/password; Bearer on all authed calls). export const LOGIN_PATH = "/auth/login"; export const LOGOUT_PATH = "/auth/logout"; @@ -123,6 +139,9 @@ export interface StreamOptions { signal?: AbortSignal; /** Timeout for opening the stream and for each quiet interval between chunks. 0 disables it. */ timeoutMs?: number; + /** HTTP method for the stream request. Default POST (body JSON-encoded); + * "GET" sends no body (dev-session downstream SSE). */ + method?: "POST" | "GET"; } export class ApiClient { @@ -222,7 +241,7 @@ export class ApiClient { body: unknown, signalOrOptions?: AbortSignal | StreamOptions, ): Promise> { - const { signal, timeoutMs } = normalizeStreamOptions(signalOrOptions); + const { signal, timeoutMs, method } = normalizeStreamOptions(signalOrOptions); // `net` only tells fetch()/the body reader to release the socket on timeout // or abort — it is never inspected to pick the error the caller sees. That // classification comes solely from raceAgainst racing the caller's own @@ -240,13 +259,13 @@ export class ApiClient { used = await this.tokens.get(); return raceAgainst( fetch(this.url(path), { - method: "POST", + method, headers: { - "Content-Type": "application/json", + ...(method === "GET" ? {} : { "Content-Type": "application/json" }), Accept: "text/event-stream", ...(await this.authHeaders(used)), }, - body: JSON.stringify(body), + ...(method === "GET" ? {} : { body: JSON.stringify(body) }), signal: net.signal, }), signal, @@ -580,13 +599,18 @@ async function toHttpError(res: Response): Promise { function normalizeStreamOptions(signalOrOptions?: AbortSignal | StreamOptions): { signal?: AbortSignal; timeoutMs: number; + method: "POST" | "GET"; } { const isSignal = !!signalOrOptions && "aborted" in signalOrOptions && "addEventListener" in signalOrOptions; const opts: StreamOptions = isSignal ? { signal: signalOrOptions as AbortSignal } : ((signalOrOptions as StreamOptions | undefined) ?? {}); - return { signal: opts.signal, timeoutMs: normalizeTimeoutMs(opts.timeoutMs ?? defaultStreamTimeoutMs()) }; + return { + signal: opts.signal, + timeoutMs: normalizeTimeoutMs(opts.timeoutMs ?? defaultStreamTimeoutMs()), + method: opts.method ?? "POST", + }; } /** Exported so tests can pin AETHER_STREAM_TIMEOUT_MS parsing without a live stream. */ diff --git a/test/brain_cloud.test.ts b/test/brain_cloud.test.ts index ed769e0..17ba1bd 100644 --- a/test/brain_cloud.test.ts +++ b/test/brain_cloud.test.ts @@ -10,9 +10,22 @@ import type { TokenStore } from "../src/core/auth.js"; const tokens = { get: async () => "aek_t" } as unknown as TokenStore; +// A LEGACY server: the dev-session route does not exist (404), so CloudBrain +// falls back to the one-way /agent/chat/stream path these tests cover. function sseFetch(events: string[]): typeof globalThis.fetch { const body = events.map((e) => `data: ${e}\n\n`).join(""); - return (async () => { + return (async (input: RequestInfo | URL) => { + const url = String(input); + if (url.includes("/agent/dev/sessions")) { + return { + ok: false, + status: 404, + headers: new Headers({ "content-type": "application/json" }), + text: async () => JSON.stringify({ detail: "Not Found" }), + json: async () => ({ detail: "Not Found" }), + body: null, + } as unknown as Response; + } const bytes = new TextEncoder().encode(body); return { ok: true, diff --git a/test/brain_cloud_dev.test.ts b/test/brain_cloud_dev.test.ts new file mode 100644 index 0000000..839cfca --- /dev/null +++ b/test/brain_cloud_dev.test.ts @@ -0,0 +1,347 @@ +// CloudBrain dev-session protocol — the bidirectional tool round-trip. +// +// Proves the spec Gate-1 client half: session create (effort + capabilities on +// the wire), tool_call frames surfacing as BrainEvents, sendToolResult POSTing +// upstream, seq-based duplicate suppression (a replayed mutating call never +// executes twice), reconnect-from-seq, control(), close() teardown, and the +// legacy fallback split (404/403 → old chat stream; other errors surface). + +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { CloudBrain } from "../src/core/brain_cloud.js"; +import { ApiClient } from "../src/core/transport.js"; +import type { BrainEvent } from "../src/core/brain_protocol.js"; +import type { TokenStore } from "../src/core/auth.js"; + +const tokens = { get: async () => "aek_t" } as unknown as TokenStore; + +interface Call { + method: string; + url: string; + body: unknown; +} + +/** A dev-protocol server fake: create → JSON; stream attempts → scripted SSE + * bodies (one per reconnect); tool-results/control/DELETE → recorded JSON. */ +function devServer(streams: string[][], opts?: { failToolResultTimes?: number }) { + const calls: Call[] = []; + let attempt = 0; + let toolResultFailures = opts?.failToolResultTimes ?? 0; + const fetchImpl = (async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + const method = init?.method ?? "GET"; + const body = init?.body ? JSON.parse(String(init.body)) : undefined; + calls.push({ method, url, body }); + + const json = (status: number, payload: unknown): Response => + ({ + ok: status < 400, + status, + headers: new Headers({ "content-type": "application/json" }), + text: async () => JSON.stringify(payload), + json: async () => payload, + body: null, + }) as unknown as Response; + + if (url.endsWith("/agent/dev/sessions") && method === "POST") { + return json(200, { + session_id: "devs_abc", + protocol_version: 1, + model: "sonnet", + tools: ["read_file", "write_file"], + }); + } + if (url.includes("/stream") && method === "GET") { + const frames = streams[Math.min(attempt, streams.length - 1)] ?? []; + attempt += 1; + const bytes = new TextEncoder().encode(frames.map((e) => `data: ${e}\n\n`).join("")); + return { + ok: true, + status: 200, + headers: new Headers({ "content-type": "text/event-stream" }), + body: (async function* (): AsyncIterable { + yield bytes; + })(), + } as unknown as Response; + } + if (url.includes("/tool-results")) { + if (toolResultFailures > 0) { + toolResultFailures -= 1; + return json(500, { detail: "transient" }); + } + return json(200, { accepted: true, duplicate: false }); + } + if (url.includes("/control")) return json(200, { ok: true, state: "running" }); + if (method === "DELETE") return json(200, {}); + return json(404, { detail: "unexpected" }); + }) as typeof globalThis.fetch; + return { fetchImpl, calls }; +} + +async function withFetch(f: typeof globalThis.fetch, run: () => Promise): Promise { + const real = globalThis.fetch; + globalThis.fetch = f; + try { + return await run(); + } finally { + globalThis.fetch = real; + } +} + +const TASK = { type: "task" as const, text: "fix it", cwd: ".", poolGb: 5, effort: "CODEPRO", model: "sonnet" }; + +function frame(obj: Record): string { + return JSON.stringify(obj); +} + +test("dev session: create carries effort + capabilities; tool_call surfaces and sendToolResult POSTs upstream", async () => { + const { fetchImpl, calls } = devServer([ + [ + frame({ type: "session", seq: 1, session_id: "devs_abc", protocol_version: 1, model: "sonnet" }), + frame({ type: "tool_call", seq: 2, tool_call_id: "tc_1", name: "read_file", args: { path: "a.py" }, risk: "read" }), + frame({ type: "tool_result_ack", seq: 3, tool_call_id: "tc_1" }), + frame({ type: "delta", seq: 4, text: "done!" }), + frame({ type: "done", seq: 5, ok: true, uvt: 10, cents: 0.1 }), + ], + ]); + await withFetch(fetchImpl, async () => { + const brain = new CloudBrain(new ApiClient("https://stub.test", tokens)); + const out: BrainEvent[] = []; + for await (const ev of brain.run(TASK)) { + out.push(ev); + if (ev.type === "tool_call") { + brain.sendToolResult(ev.id, { output: "print('hi')", exitCode: 0 }); + } + } + // give the serialized upstream POST a tick to land + await new Promise((r) => setTimeout(r, 20)); + + const create = calls.find((c) => c.url.endsWith("/agent/dev/sessions")); + assert.ok(create); + const createBody = create.body as Record; + assert.equal(createBody["effort"], "CODEPRO"); + assert.equal(createBody["surface"], "aether_agent"); + assert.ok(Array.isArray(createBody["capabilities"])); + assert.ok((createBody["capabilities"] as string[]).includes("run_shell")); + + const tc = out.find((e) => e.type === "tool_call"); + assert.ok(tc && tc.type === "tool_call"); + assert.equal(tc.id, "tc_1"); + assert.equal(tc.name, "read_file"); + assert.deepEqual(tc.args, { path: "a.py" }); + + const post = calls.find((c) => c.url.includes("/tool-results")); + assert.ok(post, "tool result was never POSTed"); + const postBody = post.body as Record; + assert.equal(postBody["tool_call_id"], "tc_1"); + assert.equal(postBody["status"], "ok"); + assert.equal(postBody["exit_code"], 0); + assert.equal(postBody["output"], "print('hi')"); + + const done = out.find((e) => e.type === "done"); + assert.ok(done && done.type === "done" && done.ok === true); + }); +}); + +test("dev session: a replayed frame (seq <= high-water mark) is skipped — a mutating tool_call never fires twice", async () => { + const tcFrame = frame({ type: "tool_call", seq: 2, tool_call_id: "tc_1", name: "write_file", args: { path: "a", content: "b" }, risk: "write" }); + const { fetchImpl } = devServer([ + [ + frame({ type: "delta", seq: 1, text: "x" }), + tcFrame, + tcFrame, // duplicate delivery of the same SSE frame + frame({ type: "done", seq: 3, ok: true, uvt: 1, cents: 0 }), + ], + ]); + await withFetch(fetchImpl, async () => { + const brain = new CloudBrain(new ApiClient("https://stub.test", tokens)); + const out: BrainEvent[] = []; + for await (const ev of brain.run(TASK)) out.push(ev); + const toolCalls = out.filter((e) => e.type === "tool_call"); + assert.equal(toolCalls.length, 1); + }); +}); + +test("dev session: a dropped stream reconnects from last_seq and finishes", async () => { + const { fetchImpl, calls } = devServer([ + [ + frame({ type: "delta", seq: 1, text: "part one" }), + // stream ends here with no terminal frame — client must reconnect + ], + [ + frame({ type: "delta", seq: 2, text: "part two" }), + frame({ type: "done", seq: 3, ok: true, uvt: 1, cents: 0 }), + ], + ]); + await withFetch(fetchImpl, async () => { + const brain = new CloudBrain(new ApiClient("https://stub.test", tokens)); + const out: BrainEvent[] = []; + for await (const ev of brain.run(TASK)) out.push(ev); + + const streamCalls = calls.filter((c) => c.url.includes("/stream")); + assert.equal(streamCalls.length, 2); + assert.ok(!streamCalls[0]!.url.includes("last_seq")); + assert.match(streamCalls[1]!.url, /last_seq=1/); + + const done = out.find((e) => e.type === "done"); + assert.ok(done && done.type === "done" && done.ok === true); + }); +}); + +test("dev session: a server error frame ends the run done ok:false (never fabricated success)", async () => { + const { fetchImpl } = devServer([ + [ + frame({ type: "delta", seq: 1, text: "partial" }), + frame({ type: "error", seq: 2, msg: "host did not return a result for run_shell within 960s" }), + ], + ]); + await withFetch(fetchImpl, async () => { + const brain = new CloudBrain(new ApiClient("https://stub.test", tokens)); + const out: BrainEvent[] = []; + for await (const ev of brain.run(TASK)) out.push(ev); + const done = out.find((e) => e.type === "done"); + assert.ok(done && done.type === "done"); + assert.equal(done.ok, false); + assert.match(done.result, /did not return a result/); + }); +}); + +test("dev session: done ok:false from the server stays ok:false", async () => { + const { fetchImpl } = devServer([ + [frame({ type: "done", seq: 1, ok: false, uvt: 1, cents: 0 })], + ]); + await withFetch(fetchImpl, async () => { + const brain = new CloudBrain(new ApiClient("https://stub.test", tokens)); + const out: BrainEvent[] = []; + for await (const ev of brain.run(TASK)) out.push(ev); + const done = out.find((e) => e.type === "done"); + assert.ok(done && done.type === "done" && done.ok === false); + }); +}); + +test("dev session: control() posts pause/steer to the control route", async () => { + const { fetchImpl, calls } = devServer([ + [frame({ type: "done", seq: 1, ok: true, uvt: 1, cents: 0 })], + ]); + await withFetch(fetchImpl, async () => { + const brain = new CloudBrain(new ApiClient("https://stub.test", tokens)); + const out: BrainEvent[] = []; + for await (const ev of brain.run(TASK)) out.push(ev); + brain.control("steer", "skip the billing code"); + await new Promise((r) => setTimeout(r, 20)); + const ctl = calls.find((c) => c.url.includes("/control")); + assert.ok(ctl, "control was never POSTed"); + const body = ctl.body as Record; + assert.equal(body["action"], "steer"); + assert.equal(body["note"], "skip the billing code"); + }); +}); + +test("dev session: close() tears the server session down (DELETE)", async () => { + const { fetchImpl, calls } = devServer([ + [frame({ type: "done", seq: 1, ok: true, uvt: 1, cents: 0 })], + ]); + await withFetch(fetchImpl, async () => { + const brain = new CloudBrain(new ApiClient("https://stub.test", tokens)); + const out: BrainEvent[] = []; + for await (const ev of brain.run(TASK)) out.push(ev); + brain.close(); + await new Promise((r) => setTimeout(r, 20)); + const del = calls.find((c) => c.method === "DELETE"); + assert.ok(del, "session was never deleted"); + assert.match(del.url, /\/agent\/dev\/sessions\/devs_abc$/); + }); +}); + +test("dev session: a transient tool-result POST failure is retried (idempotent upstream)", async () => { + const { fetchImpl, calls } = devServer( + [ + [ + frame({ type: "tool_call", seq: 1, tool_call_id: "tc_1", name: "read_file", args: { path: "a" } }), + frame({ type: "done", seq: 2, ok: true, uvt: 1, cents: 0 }), + ], + ], + { failToolResultTimes: 1 }, + ); + await withFetch(fetchImpl, async () => { + const brain = new CloudBrain(new ApiClient("https://stub.test", tokens)); + for await (const ev of brain.run(TASK)) { + if (ev.type === "tool_call") brain.sendToolResult(ev.id, { output: "x", exitCode: 0 }); + } + await new Promise((r) => setTimeout(r, 1200)); + const posts = calls.filter((c) => c.url.includes("/tool-results")); + assert.equal(posts.length, 2, "expected one failed POST and one retry"); + }); +}); + +test("legacy fallback: a 404 on session create degrades to the one-way chat stream", async () => { + const calls: Call[] = []; + const fetchImpl = (async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + calls.push({ method: init?.method ?? "GET", url, body: init?.body ? JSON.parse(String(init.body)) : undefined }); + if (url.includes("/agent/dev/sessions")) { + return { + ok: false, + status: 404, + headers: new Headers({ "content-type": "application/json" }), + text: async () => JSON.stringify({ detail: "Not Found" }), + json: async () => ({ detail: "Not Found" }), + body: null, + } as unknown as Response; + } + const bytes = new TextEncoder().encode( + [ + `data: ${frame({ type: "delta", text: "legacy reply" })}\n\n`, + `data: ${frame({ type: "done", uvt: 1, cents: 0 })}\n\n`, + ].join(""), + ); + return { + ok: true, + status: 200, + headers: new Headers({ "content-type": "text/event-stream" }), + body: (async function* (): AsyncIterable { + yield bytes; + })(), + } as unknown as Response; + }) as typeof globalThis.fetch; + + await withFetch(fetchImpl, async () => { + const brain = new CloudBrain(new ApiClient("https://stub.test", tokens)); + const out: BrainEvent[] = []; + for await (const ev of brain.run(TASK)) out.push(ev); + assert.ok(calls.some((c) => c.url.includes("/agent/chat/stream"))); + const done = out.find((e) => e.type === "done"); + assert.ok(done && done.type === "done" && done.ok === true); + // legacy path: no server session, so tool results/control are no-ops + brain.sendToolResult("tc_x", { output: "x", exitCode: 0 }); + brain.control("pause"); + await new Promise((r) => setTimeout(r, 20)); + assert.ok(!calls.some((c) => c.url.includes("/tool-results"))); + assert.ok(!calls.some((c) => c.url.includes("/control"))); + }); +}); + +test("a non-404 create failure surfaces as an error, not a silent legacy downgrade", async () => { + const fetchImpl = (async (input: RequestInfo | URL) => { + const url = String(input); + if (url.includes("/agent/dev/sessions")) { + return { + ok: false, + status: 402, + headers: new Headers({ "content-type": "application/json" }), + text: async () => JSON.stringify({ detail: "quota exhausted" }), + json: async () => ({ detail: "quota exhausted" }), + body: null, + } as unknown as Response; + } + throw new Error("legacy path must not be reached"); + }) as typeof globalThis.fetch; + + await withFetch(fetchImpl, async () => { + const brain = new CloudBrain(new ApiClient("https://stub.test", tokens)); + const out: BrainEvent[] = []; + for await (const ev of brain.run(TASK)) out.push(ev); + assert.ok(out.some((e) => e.type === "error")); + assert.ok(!out.some((e) => e.type === "monologue")); + }); +});