From e687a42565fec7e4c286fd0a05a957aeeb3d00be Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=81ngel=20Manuel=20Mart=C3=ADn?= Date: Sat, 18 Apr 2026 20:27:03 +0000 Subject: [PATCH] feat(protocol,loop): agent-owned context, operator-owned record Separate the agent's working context from the operator's canonical conversation record. The agent is free to inject scaffolding (, turn markers, nudges) without polluting what the operator persists. Protocol changes: - turn/end.messages deprecated (empty, omitted on wire) - New context/compacted notification carries summary + surviving canonical messages after compaction Compaction rewrite: - Returns CompactionResult with working_messages (for LLM) and canonical_messages (for operator) - Uses system message instead of synthetic user/assistant pair - Configurable thresholds via env vars (CONTEXT_WINDOW, COMPACTION_THRESHOLD, COMPACTION_KEEP_RECENT, MAX_COMPACTIONS) - Per-session compaction cap (default 3) Broker: context/compacted added to notification passthrough. Docs: new docs/design/context-management.md; updated SPEC, protocol, STATUS, TODO. Co-Authored-By: Claude Opus 4.6 (1M context) --- SPEC.md | 28 +++- STATUS.md | 4 +- TODO.md | 19 +++ agent/tests/e2e_supervisor_loop.rs | 109 +++++++++---- docs/design/INDEX.md | 5 + docs/design/context-management.md | 148 ++++++++++++++++++ docs/design/protocol.md | 37 ++++- loop/src/acp.rs | 141 +++++++++++------ loop/src/agentic_loop.rs | 41 ++++- loop/src/compaction.rs | 70 ++++++--- loop/src/config.rs | 42 +++++ loop/src/main.rs | 4 + loop/src/observability/langfuse.rs | 4 + loop/src/traits.rs | 19 ++- loop/tests/test_agentic_loop.rs | 21 +++ loop/tests/test_compaction.rs | 83 +++++++--- loop/tests/test_langfuse.rs | 4 + protocol/src/lib.rs | 4 +- protocol/src/messages.rs | 31 +++- protocol/src/methods.rs | 15 +- .../tests/fixtures/context_compacted.json | 8 + .../tests/fixtures/turn_end_usage_only.json | 3 + protocol/tests/wire_fixtures.rs | 14 +- server/src/tunnel/dispatch.rs | 19 ++- server/src/tunnel/run.rs | 9 +- 25 files changed, 710 insertions(+), 172 deletions(-) create mode 100644 docs/design/context-management.md create mode 100644 protocol/tests/fixtures/context_compacted.json create mode 100644 protocol/tests/fixtures/turn_end_usage_only.json diff --git a/SPEC.md b/SPEC.md index db08fd8..0569349 100644 --- a/SPEC.md +++ b/SPEC.md @@ -268,7 +268,8 @@ agent JWT in the `Authorization: Bearer` header on upgrade. | `stream/activity` | extension | agent → server | notification | broker fan-out → SSE | | `stream/toolCall` | extension | agent → server | notification | broker fan-out → SSE | | `stream/toolResult` | extension | agent → server | notification | broker fan-out → SSE | -| `turn/end` | extension | agent → server | notification | broker fan-out → SSE (operator persists) | +| `turn/end` | extension | agent → server | notification | broker fan-out → SSE (usage only; `messages` deprecated) | +| `context/compacted` | extension | agent → server | notification | broker fan-out → SSE (operator persists compacted history) | | `heartbeat` | extension | agent → server | notification | registry liveness ping | Method-name origin policy: borrow from upstream where possible. @@ -336,18 +337,37 @@ to the child process on demand. When the agent finishes a turn it emits `turn/end` as a **notification** (no response expected). The broker fans it out -to SSE subscribers; the operator is responsible for persisting the -data. +to SSE subscribers. **The `messages` field is deprecated** — agents +SHOULD send only `usage`. Operators that need per-turn persistence +should reconstruct from `stream/*` notifications or wait for +`context/compacted`. ``` Agent → server (notification): { "jsonrpc": "2.0", "method": "turn/end", "params": { - "messages": [ ...turn messages... ], "usage": { "input_tokens": 1234, "output_tokens": 567 } }} ``` +### `context/compacted` — post-compaction history sync + +Emitted after the agent compacts its working context. Carries a +prose summary of the dropped messages and the surviving canonical +messages. The operator SHOULD replace its stored history with +`messages` and record `summary` as the compaction prefix. See +[`docs/design/context-management.md`](./docs/design/context-management.md). + +``` +Agent → server (notification): + { "jsonrpc": "2.0", "method": "context/compacted", + "params": { + "summary": "...", + "messages": [ ...surviving recent messages... ], + "usage": { "input_tokens": 1200, "output_tokens": 400 } + }} +``` + ## REST surface Agent-facing routes authenticate with agent JWTs (scoped to the diff --git a/STATUS.md b/STATUS.md index 7a3d2a3..fcdd434 100644 --- a/STATUS.md +++ b/STATUS.md @@ -71,9 +71,9 @@ broker. The code is mid-migration on `refactor/stateless-broker`: | Crate | State | |---|---| -| `overloop` | Migrated to the broker wire shape. Depends on `overacp-protocol` for method-name constants and for payload types (`TextDelta`, `Activity`, `TurnEndParams`, `QuotaUpdateRequest`, `SessionMessageParams`) — outbound notifications are constructed from the typed structs and inbound `session/message` frames are deserialized into `SessionMessageParams` before being converted to the LLM-facing `Message` shape. (`llm::Message` stays local because it carries typed `tool_calls: Vec` that the LLM client needs; the bridge to the wire happens via a serde round-trip inside `AcpClient::turn_end`.) `AcpClient` now speaks: single cold-start `initialize` (no more per-message re-init), push-based `next_push()` that blocks on stdin until a `session/message` or `session/cancel` notification arrives, fire-and-forget `turn_end` notification, plus the streaming / heartbeat / quota methods from before. `agentic_loop::run` no longer polls for new messages inside the turn — the outer loop in `main.rs` owns message delivery. `poll/newMessages` and `turn/save` are gone. Tool-call streaming (`stream/toolCall` before every invocation, `stream/toolResult` after, both echoing the same `id` so SSE consumers can pair them) is now emitted by `agentic_loop::run`; failing tool invocations set `is_error = true` on the result frame. Optional Sentry error reporting landed behind cargo feature `sentry` (off by default, zero overhead when unused). New env vars: `OVERLOOP_AGENT_NAME` (process-local fleet identity, attached to the root tracing span and to Sentry events as a tag), `SENTRY_DSN`, `SENTRY_ENVIRONMENT` (default `local`), `SENTRY_TRACES_SAMPLE_RATE` (default `0.1`). | +| `overloop` | Migrated to the broker wire shape. Depends on `overacp-protocol` for method-name constants and for payload types (`TextDelta`, `Activity`, `TurnEndParams`, `QuotaUpdateRequest`, `SessionMessageParams`) — outbound notifications are constructed from the typed structs and inbound `session/message` frames are deserialized into `SessionMessageParams` before being converted to the LLM-facing `Message` shape. (`llm::Message` stays local because it carries typed `tool_calls: Vec` that the LLM client needs; the bridge to the wire happens via a serde round-trip inside `AcpClient::turn_end`.) `AcpClient` now speaks: single cold-start `initialize` (no more per-message re-init), push-based `next_push()` that blocks on stdin until a `session/message` or `session/cancel` notification arrives, fire-and-forget `turn_end` notification, plus the streaming / heartbeat / quota methods from before. `agentic_loop::run` no longer polls for new messages inside the turn — the outer loop in `main.rs` owns message delivery. `poll/newMessages` and `turn/save` are gone. Tool-call streaming (`stream/toolCall` before every invocation, `stream/toolResult` after, both echoing the same `id` so SSE consumers can pair them) is now emitted by `agentic_loop::run`; failing tool invocations set `is_error = true` on the result frame. Optional Sentry error reporting landed behind cargo feature `sentry` (off by default, zero overhead when unused). New env vars: `OVERLOOP_AGENT_NAME` (process-local fleet identity, attached to the root tracing span and to Sentry events as a tag), `SENTRY_DSN`, `SENTRY_ENVIRONMENT` (default `local`), `SENTRY_TRACES_SAMPLE_RATE` (default `0.1`). Auto-compaction: threshold-triggered LLM summarization (default 80% of 128K context window, keeps last 10 messages, max 3 rounds per session). After compaction, emits `context/compacted` notification carrying the surviving canonical messages and a prose summary; `turn/end` no longer carries messages (deprecated). Configurable via `CONTEXT_WINDOW`, `COMPACTION_THRESHOLD`, `COMPACTION_KEEP_RECENT`, `MAX_COMPACTIONS`. See [`docs/design/context-management.md`](./docs/design/context-management.md). | | `overacp-compute-core` | Landed as a standalone library. `ComputeProvider` trait + node/exec/log types + `${provider:path:key}` config resolver. The broker no longer depends on it; operators can pull it in directly. | -| `overacp-protocol` | Landed as a workspace member. Carries the canonical `Claims` shape (`{sub, role, user?, exp, iss}` — matches the broker's `server/src/auth.rs`), method-name constants (including `TURN_END`, `SESSION_CANCEL`, `STREAM_TOOL_CALL`, `STREAM_TOOL_RESULT`), and payload types for every request / notification on the tunnel (`InitializeResponse`, `SessionMessageParams`, `SessionCancelParams`, `TurnEndParams`, stream params, quota params). JWT helpers (`mint_token`, `validate_token`, `peek_claims_unverified`) sit on top of `jsonwebtoken`. Pure types, no tokio. 23 unit + fixture tests (10 JWT round-trips, 13 wire payload round-trips including a `Content::Blocks` multimodal fixture). Consumed by `overloop` as of Phase 1; `overacp-agent` and `overacp-server` follow in later phases. | +| `overacp-protocol` | Landed as a workspace member. Carries the canonical `Claims` shape (`{sub, role, user?, exp, iss}` — matches the broker's `server/src/auth.rs`), method-name constants (including `TURN_END`, `CONTEXT_COMPACTED`, `SESSION_CANCEL`, `STREAM_TOOL_CALL`, `STREAM_TOOL_RESULT`), and payload types for every request / notification on the tunnel (`InitializeResponse`, `SessionMessageParams`, `SessionCancelParams`, `TurnEndParams`, `ContextCompactedParams`, stream params, quota params). JWT helpers (`mint_token`, `validate_token`, `peek_claims_unverified`) sit on top of `jsonwebtoken`. Pure types, no tokio. 23 unit + fixture tests (10 JWT round-trips, 13 wire payload round-trips including a `Content::Blocks` multimodal fixture). Consumed by `overloop` as of Phase 1; `overacp-agent` and `overacp-server` follow in later phases. | | `overacp-agent` | Supervisor + stdio bridge landed (cherry-picked from `feat/0.2-protocol-and-0.3-agent` commits `5e57978` + `e001b7d`). Crate modules: `config` (env-var config: `OVERACP_TOKEN`, `OVERACP_SERVER_URL`, `OVERACP_WORKSPACE`, `OVERACP_AGENT_BINARY`), `tunnel` (tokio-tungstenite WebSocket client with exponential backoff 1s → 30s and retry-forever semantics), `bridge` (content-agnostic byte pump: WS text frames ↔ child stdio, no JSON parsing, runs under `tokio::select!` so either side closing terminates the pair), `process` (child spawn helpers with stdin/stdout piped and stderr inherited, including graceful wait + kill-and-reap on drop), `adapter` (`AgentAdapter` trait + `LoopAdapter` impl — the only adapter in 0.3), `workspace` (`WorkspaceSync` trait + `NoopSync` default), `run` (top-level orchestration: pull workspace, peek JWT `sub` for the tunnel URL, connect with backoff, spawn child, pump bridge, reap child, push workspace). 25 unit tests: tunnel URL rewriting (http→ws, https→wss, ws passthrough, port preservation), `LoopAdapter::command` env-var / reusability, backoff schedule + cap, connect error paths (unreachable host, malformed URL), `process::{spawn, write_line, read_line}` with newline handling + EOF + missing-binary error, `NoopSync` pull/push, `bridge::{ws_to_stdin, stdout_to_ws}` over generic in-memory streams (text frame forwarding, Close frame → TunnelClosed, ping/binary skip, ws-read error, EOF → ProcessExited, sink-failure → Error), and `run::reap_child` for the AlreadyExited + Killed branches. Plus a hermetic integration test `e2e_supervisor_loop.rs` that spins up the broker in-process, a wiremock mock LLM, and launches a real `overacp-agent` subprocess hosting a real `overloop` child — POSTs a user message via REST, subscribes to SSE, and asserts a `turn/end` frame with nonzero usage arrives within 30s. Planned `claude-code` and `codex` adapters remain deferred per TODO § 0.3. | | `overacp-server` | Broker refactor complete on `refactor/stateless-broker`: JWT `Claims` + `mint`, operator hooks (`BootProvider`, `ToolHost`, `QuotaPolicy`) with default impls, hook-delegating tunnel dispatch, `AgentRegistry`, `MessageQueue`, and the new REST surface (`POST /tokens`, `GET /agents`, `GET /agents/{id}`, `DELETE /agents/{id}`, `POST /agents/{id}/messages`, `GET /agents/{id}/stream`, `POST /agents/{id}/cancel`) with JWT middleware gating. The legacy `SessionStore`, compute REST surface, and HTTP Basic auth have been removed. | | `overacp-tools-mcp` | Not started. | diff --git a/TODO.md b/TODO.md index f508c95..8e39095 100644 --- a/TODO.md +++ b/TODO.md @@ -81,6 +81,25 @@ crate landed. All items block calling the loop "protocol-conformant". deployment adapters can supply tool descriptors without touching the supervisor. +## 0.3.y — context management + +See [`docs/design/context-management.md`](./docs/design/context-management.md). + +- [x] Auto-compaction at configurable threshold (default 80% of 128K). + LLM summarizes older messages; agent injects + `` system message into working context. +- [x] `context/compacted` notification — emits summary + canonical + surviving messages to operator for persistence. +- [x] `turn/end` messages field deprecated — agents send usage only. +- [x] Configurable thresholds via env vars: `CONTEXT_WINDOW`, + `COMPACTION_THRESHOLD`, `COMPACTION_KEEP_RECENT`, + `MAX_COMPACTIONS`. +- [x] Per-session compaction cap (default 3). +- [ ] Optional `memory_flush` tool hook — invoke an operator-provided + tool before summarizing so long-term facts can be extracted. +- [ ] Live turn marker — inject `` system message before + current user input. Agent-internal, no protocol change. + ## 0.4 — `overacp-server`: stateless message broker (landed) The centerpiece. A small HTTP + WebSocket router that terminates diff --git a/agent/tests/e2e_supervisor_loop.rs b/agent/tests/e2e_supervisor_loop.rs index 219130e..5bd129b 100644 --- a/agent/tests/e2e_supervisor_loop.rs +++ b/agent/tests/e2e_supervisor_loop.rs @@ -71,10 +71,13 @@ fn ensure_workspace_built() { // ── Shared pipeline helpers ──────────────────────────────────────── -/// Outcome of a full-stack pipeline run: the `turn/end` SSE frame -/// plus the KillOnDrop guard (kept alive until assertions finish). +/// Outcome of a full-stack pipeline run: the `turn/end` SSE frame, +/// all collected SSE frames, and the KillOnDrop guard (kept alive +/// until assertions finish). struct PipelineRun { frame: serde_json::Value, + /// All SSE frames received before (and including) `turn/end`. + all_frames: Vec, _supervisor: KillOnDrop, } @@ -136,32 +139,52 @@ async fn run_pipeline( .unwrap(); assert!(push_resp.status().is_success()); - // Wait for turn/end. - let frame = match timeout(Duration::from_secs(30), find_turn_end(&mut sse_stream)).await { - Ok(Ok(Some(v))) => v, - Ok(Ok(None)) => panic!("SSE closed without turn/end"), - Ok(Err(e)) => panic!("SSE error: {e}"), - Err(_) => panic!("timeout waiting for turn/end after 30s"), - }; + // Wait for turn/end, collecting all SSE frames along the way. + let (frame, all_frames) = + match timeout(Duration::from_secs(30), find_turn_end(&mut sse_stream)).await { + Ok(Ok(Some(v))) => v, + Ok(Ok(None)) => panic!("SSE closed without turn/end"), + Ok(Err(e)) => panic!("SSE error: {e}"), + Err(_) => panic!("timeout waiting for turn/end after 30s"), + }; assert_eq!(frame["method"], "turn/end"); PipelineRun { frame, + all_frames, _supervisor: supervisor, } } -/// Check that `messages` contains a message with the given role -/// whose content includes `needle`. -fn assert_message_contains(messages: &[serde_json::Value], role: &str, needle: &str, label: &str) { - let found = messages.iter().any(|m| { - m["role"] == role - && m["content"] - .as_str() - .map(|s| s.contains(needle)) - .unwrap_or(false) +/// Check that streamed SSE frames contain a `stream/textDelta` whose +/// text includes `needle`, or a `stream/toolResult` whose content +/// includes `needle`. +fn assert_streamed_contains(frames: &[serde_json::Value], needle: &str, label: &str) { + let found = frames.iter().any(|f| { + // Check stream/textDelta + if f["method"] == "stream/textDelta" { + if let Some(text) = f["params"]["text"].as_str() { + if text.contains(needle) { + return true; + } + } + } + // Check stream/toolResult + if f["method"] == "stream/toolResult" { + let content = &f["params"]["content"]; + if let Some(s) = content.as_str() { + if s.contains(needle) { + return true; + } + } + // content may be an object or array — check stringified + if content.to_string().contains(needle) { + return true; + } + } + false }); - assert!(found, "{label}, got: {messages:?}"); + assert!(found, "{label}, frames: {frames:?}"); } /// Poll `GET /agents/{id}` until the tunnel is connected. Gives up @@ -192,13 +215,16 @@ async fn wait_for_connected( false } -/// Drain an SSE byte-stream and return the first parsed JSON frame -/// whose `method` field is `"turn/end"`. -async fn find_turn_end(stream: &mut S) -> Result, reqwest::Error> +/// Drain an SSE byte-stream, collecting all parsed JSON frames, and +/// return the `turn/end` frame plus all frames seen so far. +async fn find_turn_end( + stream: &mut S, +) -> Result)>, reqwest::Error> where S: Stream> + Unpin, { let mut buf = String::new(); + let mut all_frames = Vec::new(); while let Some(chunk) = stream.next().await { let chunk: Bytes = chunk?; match str::from_utf8(&chunk) { @@ -214,8 +240,11 @@ where if let Some(json_str) = line.strip_prefix("data: ") { if let Ok(value) = serde_json::from_str::(json_str) { if value["method"] == "turn/end" { - return Ok(Some(value)); + let turn_end = value.clone(); + all_frames.push(value); + return Ok(Some((turn_end, all_frames))); } + all_frames.push(value); } } } @@ -243,7 +272,7 @@ async fn full_stack_turn_end_over_real_tunnel() { ) .await; - let messages = run.frame["params"]["messages"].as_array().unwrap(); + // turn/end carries usage only (messages deprecated). let input_tokens = run.frame["params"]["usage"]["input_tokens"] .as_u64() .unwrap_or(0); @@ -252,7 +281,8 @@ async fn full_stack_turn_end_over_real_tunnel() { .unwrap_or(0); assert!(input_tokens > 0, "input_tokens not reported"); assert!(output_tokens > 0, "output_tokens not reported"); - assert_message_contains(messages, "assistant", "HELLO", "expected HELLO"); + // Verify the agent's response arrived via stream/textDelta. + assert_streamed_contains(&run.all_frames, "HELLO", "expected HELLO in stream"); } // ── Operator-provided tool e2e ────────────────────────────────────── @@ -314,13 +344,16 @@ async fn full_stack_operator_tool_round_trip() { ) .await; - let messages = run.frame["params"]["messages"].as_array().unwrap(); - assert_message_contains(messages, "tool", "sunny", "expected weather tool result"); - assert_message_contains( - messages, - "assistant", + // Verify tool result and final text arrived via stream/*. + assert_streamed_contains( + &run.all_frames, + "sunny", + "expected weather tool result in stream", + ); + assert_streamed_contains( + &run.all_frames, "WEATHER_RESULT", - "expected final text", + "expected final text in stream", ); } @@ -350,7 +383,15 @@ async fn full_stack_builtin_tool_round_trip() { ) .await; - let messages = run.frame["params"]["messages"].as_array().unwrap(); - assert_message_contains(messages, "tool", "CANARY_VALUE", "expected marker content"); - assert_message_contains(messages, "assistant", "BUILTIN_OK", "expected final text"); + // Verify tool result and final text arrived via stream/*. + assert_streamed_contains( + &run.all_frames, + "CANARY_VALUE", + "expected marker content in stream", + ); + assert_streamed_contains( + &run.all_frames, + "BUILTIN_OK", + "expected final text in stream", + ); } diff --git a/docs/design/INDEX.md b/docs/design/INDEX.md index 1443420..db72714 100644 --- a/docs/design/INDEX.md +++ b/docs/design/INDEX.md @@ -18,6 +18,11 @@ For the high-level architecture and milestone roadmap see the Kafka-Connect-shaped REST API for managing compute pools and agent lifecycle. Replaced by the stateless message broker model in [`SPEC.md`](../../SPEC.md). Kept for context only. +- [`context-management.md`](./context-management.md) — how + over/ACP separates the agent's working context from the operator's + canonical conversation record. Covers auto-compaction, the + `context/compacted` notification, agent-internal scaffolding, and + the optional memory-flush hook. - [`workspace-sync.md`](./workspace-sync.md) — pluggable workspace hydration and persistence: where it lives in the architecture (agent supervisor, not the broker), the `WorkspaceSync` trait, diff --git a/docs/design/context-management.md b/docs/design/context-management.md new file mode 100644 index 0000000..68d055e --- /dev/null +++ b/docs/design/context-management.md @@ -0,0 +1,148 @@ +# Context management + +How over/ACP separates the agent's working context from the +operator's canonical conversation record. + +## Principle + +The agent owns its **working context** — the message array it feeds +the LLM. The operator owns the **canonical record** — the +authoritative history of what was actually said. + +The boundary between them is the protocol: + +- `initialize` (operator → agent): the operator seeds the working + context with whatever history window it chooses. +- `stream/*` (agent → operator): the operator observes every turn in + real time. +- `context/compacted` (agent → operator): the agent tells the + operator "I compacted; here is the surviving canonical history and + a summary of what was dropped — please persist." +- `turn/end` (agent → operator): a lightweight turn-complete signal + carrying `usage` only (the `messages` field is deprecated). + +The agent is free to inject any scaffolding it wants into its working +context — compaction summaries, turn markers, wind-down warnings, +silence nudges — without those synthetic messages reaching the +operator's store. + +## Auto-compaction + +overloop monitors estimated token usage on every loop iteration. When +the working context exceeds a configurable threshold of the context +window, the agent compacts: + +1. **Split** — keep the system prompt and the last N messages + (default: 10). Everything in between is the compaction target. +2. **Summarize** — an LLM call reduces the compaction target to a + prose summary preserving decisions, file changes, findings, task + state, and commitments. +3. **Rebuild** — the working context becomes: + `[system_prompt, summary, ...recent]`. + The `` message is `Role::System` — it tells the + LLM that prior context was summarized. +4. **Notify** — the agent emits `context/compacted` carrying the + summary and the canonical (non-synthetic) recent messages so the + operator can update its store. +5. **Cap** — a per-session counter limits compaction rounds (default: + 3). After the cap, the agent stops compacting and lets the LLM hit + a natural length limit. + +### Configuration + +All thresholds are environment variables with sensible defaults: + +| Env var | Default | Description | +|--------------------------|-----------|-------------| +| `CONTEXT_WINDOW` | `128000` | Total context window in tokens | +| `COMPACTION_THRESHOLD` | `0.80` | Fraction of window that triggers compaction | +| `COMPACTION_KEEP_RECENT` | `10` | Messages preserved during compaction | +| `MAX_COMPACTIONS` | `3` | Max compaction rounds per session | + +### Token estimation + +Token counts are estimated locally using a `chars / 4` heuristic for +text and a flat 765 tokens per image block. No external tokenizer +dependency. + +## `context/compacted` notification + +```jsonc +{ + "jsonrpc": "2.0", + "method": "context/compacted", + "params": { + "summary": "User asked to refactor auth middleware. We identified ...", + "messages": [ + { "role": "user", "content": "now deploy it" }, + { "role": "assistant", "content": "Deploying to staging..." } + ], + "usage": { "input_tokens": 1200, "output_tokens": 400 } + } +} +``` + +The operator SHOULD: +- Replace its stored message history for this conversation with + `messages`. +- Store `summary` as the compaction prefix so that a future + `BootProvider::initialize` can return both the summary and the + surviving messages. + +## `turn/end` (updated) + +`turn/end` is now a lightweight completion signal: + +```jsonc +{ + "jsonrpc": "2.0", + "method": "turn/end", + "params": { + "usage": { "input_tokens": 5000, "output_tokens": 1200 } + } +} +``` + +The `messages` field is deprecated and omitted by conforming agents. +Operators that need per-turn message persistence should reconstruct +from `stream/*` notifications or wait for `context/compacted`. + +## Interaction with `initialize` + +On cold start, `BootProvider::initialize` returns the system prompt +and a curated history window. The agent loads this into its working +context and immediately layers on any scaffolding it needs — turn +markers, nudges, etc. Those injections are ephemeral; they exist +only in the agent's in-memory message array. + +If the operator previously received a `context/compacted` +notification, it can include the `summary` as a system message at +the front of the returned history so the agent starts with prior +context already folded in. + +## Agent-internal scaffolding + +The following are injected into the working context but never flow +to the operator: + +- **`` system message** — the LLM-generated + summary of dropped history, wrapped in a sentinel tag. +- **Wind-down warning** — a system message injected when the + iteration budget is nearly exhausted. +- **Silence nudge** — injected when the LLM produces no output for + several iterations. +- **Turn markers** (future) — a `` system message injected + before the current user input to delineate prior history from the + current request. Not yet implemented; can be added as a follow-up + without protocol changes. + +## Optional memory-flush hook (planned) + +Before summarizing, the compaction path can check for an +operator-provided `memory_flush` tool. If registered, the agent +invokes it with the to-compact messages so the operator can extract +long-term facts (user preferences, architectural decisions, etc.) +to durable storage before the messages are summarized away. + +This is not yet implemented; the hook point is marked with a TODO +in `loop/src/compaction.rs`. diff --git a/docs/design/protocol.md b/docs/design/protocol.md index 3dbba5c..554ce46 100644 --- a/docs/design/protocol.md +++ b/docs/design/protocol.md @@ -149,6 +149,7 @@ Each method is tagged with its **origin**: | `quota/check` | extension | agent → server | request | | `quota/update` | extension | agent → server | request | | `turn/end` | extension | agent → server | notification | +| `context/compacted` | extension | agent → server | notification | | `stream/textDelta` | extension | agent → server | notification | | `stream/activity` | extension | agent → server | notification | | `stream/toolCall` | extension | agent → server | notification | @@ -206,21 +207,43 @@ hooks. No upstream standard covers them, so the names are ours. `turn/end` is a **fire-and-forget notification** that an agent emits when it finishes a turn. The broker fans it out to SSE -subscribers; persisting the turn data is the operator's job (the -operator's SSE consumer reads `turn/end` frames out of -`/agents/{id}/stream` and writes them to whatever store it owns). +subscribers. **The `messages` field is deprecated** — agents SHOULD +send only `usage`. Operators that need per-turn message persistence +should reconstruct from `stream/*` notifications or wait for +`context/compacted`. ```jsonc -// turn/end notification +// turn/end notification (messages deprecated, omitted) { "jsonrpc": "2.0", "method": "turn/end", "params": { + "usage": { "input_tokens": 100, "output_tokens": 50 } + } +} +``` + +`context/compacted` is a **fire-and-forget notification** emitted +after the agent compacts its working context. It carries a prose +`summary` of the dropped messages and the surviving canonical +`messages`. The operator SHOULD replace its stored history with +`messages` and record `summary` as the compaction prefix so that a +future `BootProvider::initialize` can return both. See +[`context-management.md`](./context-management.md) for the full +design. + +```jsonc +// context/compacted notification +{ + "jsonrpc": "2.0", + "method": "context/compacted", + "params": { + "summary": "User asked to refactor auth...", "messages": [ - { "role": "user", "content": "what's the weather?" }, - { "role": "assistant", "content": "I'll check." } + { "role": "user", "content": "now deploy it" }, + { "role": "assistant", "content": "Deploying..." } ], - "usage": { "input_tokens": 100, "output_tokens": 50 } + "usage": { "input_tokens": 1200, "output_tokens": 400 } } } ``` diff --git a/loop/src/acp.rs b/loop/src/acp.rs index 4184684..0f815c2 100644 --- a/loop/src/acp.rs +++ b/loop/src/acp.rs @@ -7,8 +7,9 @@ use anyhow::{Context, Result}; use overacp_protocol::messages::{ - Activity, Message as ProtoMessage, QuotaUpdateRequest, Role as ProtoRole, SessionMessageParams, - TextDelta, ToolCallNotification, ToolResultNotification, TurnEndParams, Usage as ProtoUsage, + Activity, ContextCompactedParams, Message as ProtoMessage, QuotaUpdateRequest, + Role as ProtoRole, SessionMessageParams, TextDelta, ToolCallNotification, + ToolResultNotification, TurnEndParams, Usage as ProtoUsage, }; use overacp_protocol::methods; use serde::{Deserialize, Serialize}; @@ -247,25 +248,42 @@ impl AcpService for AcpClient { ) } - fn turn_end(&mut self, messages: &[Message], usage: &Usage) -> Result<()> { - // Convert LLM-facing messages into the protocol's `Message` - // shape via serde round-trip. The two types serialize - // identically (same field names, `tool_calls` as Value on the - // protocol side absorbs the richer ToolCall typing); using - // the typed `TurnEndParams` here makes the broker's wire - // contract enforce the payload shape at compile time rather - // than at runtime when the broker rejects it. + fn turn_end(&mut self, _messages: &[Message], usage: &Usage) -> Result<()> { + // messages is deprecated — agents send only usage. The + // operator reconstructs the turn from stream/* notifications; + // the authoritative post-compaction snapshot comes via + // context/compacted. + let params = TurnEndParams { + messages: vec![], + usage: ProtoUsage { + input_tokens: usage.prompt_tokens, + output_tokens: usage.completion_tokens, + }, + }; + self.send_notification(methods::TURN_END, Some(serde_json::to_value(params)?)) + } + + fn context_compacted( + &mut self, + summary: &str, + messages: &[Message], + usage: &Usage, + ) -> Result<()> { let proto_messages: Vec = serde_json::from_value(serde_json::to_value(messages)?) - .context("convert llm::Message to protocol::Message for turn/end")?; - let params = TurnEndParams { + .context("convert llm::Message to protocol::Message for context/compacted")?; + let params = ContextCompactedParams { + summary: summary.to_string(), messages: proto_messages, usage: ProtoUsage { input_tokens: usage.prompt_tokens, output_tokens: usage.completion_tokens, }, }; - self.send_notification(methods::TURN_END, Some(serde_json::to_value(params)?)) + self.send_notification( + methods::CONTEXT_COMPACTED, + Some(serde_json::to_value(params)?), + ) } fn quota_check(&mut self) -> Result { @@ -604,7 +622,7 @@ mod tests { } #[test] - fn turn_end_roundtrips_anthropic_image_blocks() { + fn turn_end_sends_no_messages() { use crate::llm::{ContentBlock, TypedBlock}; let mut acp = mock_acp(""); let messages = vec![Message { @@ -630,11 +648,12 @@ mod tests { let out = String::from_utf8(acp.writer.clone()).unwrap(); let parsed: Value = serde_json::from_str(out.trim()).unwrap(); - let content = &parsed["params"]["messages"][0]["content"]; - assert_eq!(content.as_array().unwrap().len(), 2); - assert_eq!(content[0]["type"], "text"); - assert_eq!(content[1]["type"], "image"); - assert_eq!(content[1]["source"]["media_type"], "image/png"); + // messages is deprecated — should be absent (skip_serializing_if = "Vec::is_empty") + assert!( + parsed["params"].get("messages").is_none(), + "turn/end should not send messages (deprecated)" + ); + assert_eq!(parsed["params"]["usage"]["input_tokens"], 10); } #[test] @@ -660,19 +679,11 @@ mod tests { } #[test] - fn turn_end_roundtrips_multimodal_content_blocks() { - use crate::llm::{ContentBlock, TypedBlock}; + fn turn_end_usage_only() { let mut acp = mock_acp(""); let messages = vec![Message { - role: Role::User, - content: Some(Content::Blocks(vec![ - TypedBlock::Known(ContentBlock::Text { - text: "what's this?".into(), - }), - TypedBlock::Known(ContentBlock::ImageUrl { - image_url: json!({"url": "data:image/png;base64,abc"}), - }), - ])), + role: Role::Assistant, + content: Some(Content::Text("done".into())), tool_calls: None, tool_call_id: None, }]; @@ -686,39 +697,65 @@ mod tests { let out = String::from_utf8(acp.writer.clone()).unwrap(); let parsed: Value = serde_json::from_str(out.trim()).unwrap(); - let content = &parsed["params"]["messages"][0]["content"]; - assert!(content.is_array()); - assert_eq!(content.as_array().unwrap().len(), 2); - assert_eq!(content[0]["type"], "text"); - assert_eq!(content[0]["text"], "what's this?"); - assert_eq!(content[1]["type"], "image_url"); - assert_eq!(content[1]["image_url"]["url"], "data:image/png;base64,abc"); + assert_eq!(parsed["method"], "turn/end"); + assert!(parsed.get("id").is_none() || parsed["id"].is_null()); + assert_eq!(parsed["params"]["usage"]["input_tokens"], 100); + assert_eq!(parsed["params"]["usage"]["output_tokens"], 50); + // messages deprecated — absent + assert!(parsed["params"].get("messages").is_none()); } #[test] - fn turn_end_wraps_messages_and_maps_usage_field_names() { + fn context_compacted_roundtrips_messages() { + use crate::llm::{ContentBlock, TypedBlock}; let mut acp = mock_acp(""); - let messages = vec![Message { - role: Role::Assistant, - content: Some(Content::Text("done".into())), - tool_calls: None, - tool_call_id: None, - }]; + let messages = vec![ + Message { + role: Role::User, + content: Some(Content::Text("deploy it".into())), + tool_calls: None, + tool_call_id: None, + }, + Message { + role: Role::User, + content: Some(Content::Blocks(vec![ + TypedBlock::Known(ContentBlock::Text { + text: "with this image".into(), + }), + TypedBlock::Known(ContentBlock::ImageUrl { + image_url: json!({"url": "data:image/png;base64,abc"}), + }), + ])), + tool_calls: None, + tool_call_id: None, + }, + ]; let usage = Usage { - prompt_tokens: 100, - completion_tokens: 50, - total_tokens: 150, + prompt_tokens: 200, + completion_tokens: 80, + total_tokens: 280, ..Default::default() }; - acp.turn_end(&messages, &usage).unwrap(); + acp.context_compacted("summary of earlier work", &messages, &usage) + .unwrap(); let out = String::from_utf8(acp.writer.clone()).unwrap(); let parsed: Value = serde_json::from_str(out.trim()).unwrap(); - assert_eq!(parsed["method"], "turn/end"); + assert_eq!(parsed["method"], "context/compacted"); assert!(parsed.get("id").is_none() || parsed["id"].is_null()); - assert_eq!(parsed["params"]["usage"]["input_tokens"], 100); - assert_eq!(parsed["params"]["usage"]["output_tokens"], 50); - assert_eq!(parsed["params"]["messages"][0]["role"], "assistant"); + assert_eq!(parsed["params"]["summary"], "summary of earlier work"); + assert_eq!(parsed["params"]["usage"]["input_tokens"], 200); + assert_eq!(parsed["params"]["usage"]["output_tokens"], 80); + + let msgs = parsed["params"]["messages"].as_array().unwrap(); + assert_eq!(msgs.len(), 2); + assert_eq!(msgs[0]["role"], "user"); + assert_eq!(msgs[0]["content"], "deploy it"); + // Second message has multimodal blocks + let blocks = msgs[1]["content"].as_array().unwrap(); + assert_eq!(blocks.len(), 2); + assert_eq!(blocks[0]["type"], "text"); + assert_eq!(blocks[1]["type"], "image_url"); } #[test] diff --git a/loop/src/agentic_loop.rs b/loop/src/agentic_loop.rs index 22680e9..6c835c1 100644 --- a/loop/src/agentic_loop.rs +++ b/loop/src/agentic_loop.rs @@ -18,8 +18,6 @@ use crate::traits::{AcpService, LlmService}; /// Maximum characters captured for tool input/output in a Langfuse span. const LF_TRUNCATE: usize = 4096; -const CONTEXT_WINDOW: usize = 128_000; -const COMPACTION_THRESHOLD: f64 = 0.80; const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(30); const WIND_DOWN_ITERS: usize = 5; @@ -32,6 +30,18 @@ pub struct LoopConfig { /// Model name recorded on Langfuse generation spans. Free-form label; /// the trace is untouched when the `SessionTrace` is disabled. pub model: String, + /// Total context window in tokens. Defaults to 128,000. + pub context_window: usize, + /// Fraction of `context_window` at which auto-compaction triggers. + /// Defaults to 0.80. + pub compaction_threshold: f64, + /// Number of most-recent messages to preserve during compaction. + /// Defaults to 10. + pub compaction_keep_recent: usize, + /// Maximum number of compaction rounds per session. After this + /// limit the agent stops compacting and lets the LLM hit a + /// natural length limit. Defaults to 3. + pub max_compactions: usize, /// When true, attach a redacted chat-log snapshot to Langfuse /// generation `input`. Off by default. pub langfuse_capture_input: bool, @@ -51,6 +61,7 @@ pub async fn run( let mut last_heartbeat = Instant::now(); let mut response_text = String::new(); let mut tool_count: usize = 0; + let mut compaction_count: usize = 0; for iteration in 0..config.max_iterations { // Timeout check @@ -74,12 +85,30 @@ pub async fn run( // Compaction check let est_tokens = estimate_tokens(messages); - if est_tokens as f64 > CONTEXT_WINDOW as f64 * COMPACTION_THRESHOLD { + let threshold = config.context_window as f64 * config.compaction_threshold; + if est_tokens as f64 > threshold && compaction_count < config.max_compactions { info!( - "Context at ~{}% — compacting", - est_tokens * 100 / CONTEXT_WINDOW + "Context at ~{}% — compacting ({}/{})", + est_tokens * 100 / config.context_window, + compaction_count + 1, + config.max_compactions, + ); + let result = compact_messages(llm, messages, config.compaction_keep_recent).await?; + if !result.summary.is_empty() { + if let Err(e) = + acp.context_compacted(&result.summary, &result.canonical_messages, &total_usage) + { + error!("Failed to emit context/compacted: {}", e); + } + compaction_count += 1; + } + *messages = result.working_messages; + } else if est_tokens as f64 > threshold { + warn!( + "Context at ~{}% but compaction cap ({}) reached — skipping", + est_tokens * 100 / config.context_window, + config.max_compactions, ); - *messages = compact_messages(llm, messages, 10).await?; } // Wind-down warning diff --git a/loop/src/compaction.rs b/loop/src/compaction.rs index 26227a7..bc2d908 100644 --- a/loop/src/compaction.rs +++ b/loop/src/compaction.rs @@ -12,17 +12,38 @@ Preserve all important context: decisions made, files modified, \ key findings, current task state, and any commitments or next steps. \ Be concise but complete — this summary replaces the original messages."; +/// Result of a compaction operation, separating what the LLM sees +/// from what the operator should persist. +pub struct CompactionResult { + /// Messages for the LLM's working context. Includes a + /// `` system message summarizing dropped + /// history — this is agent-internal scaffolding. + pub working_messages: Vec, + /// Clean recent messages for the operator (no scaffolding). + /// The operator should persist these as the new canonical + /// conversation history. + pub canonical_messages: Vec, + /// Prose summary of the messages that were dropped. + pub summary: String, +} + /// Compact older messages into a summary when context is too large. /// /// Keeps the system prompt and the last `keep_recent` messages intact. -/// Summarizes everything in between using an LLM call. +/// Summarizes everything in between using an LLM call, then returns +/// a [`CompactionResult`] with both the LLM-facing working context +/// and the clean canonical messages for operator persistence. pub async fn compact_messages( llm: &(impl LlmService + ?Sized), messages: &[Message], keep_recent: usize, -) -> Result> { +) -> Result { if messages.len() <= keep_recent + 2 { - return Ok(messages.to_vec()); + return Ok(CompactionResult { + working_messages: messages.to_vec(), + canonical_messages: messages.to_vec(), + summary: String::new(), + }); } let system_msg = messages.first().filter(|m| m.role == Role::System).cloned(); @@ -32,12 +53,20 @@ pub async fn compact_messages( let split_point = find_safe_split(messages, start, proposed); if split_point <= start { - return Ok(messages.to_vec()); + return Ok(CompactionResult { + working_messages: messages.to_vec(), + canonical_messages: messages.to_vec(), + summary: String::new(), + }); } let to_compact = &messages[start..split_point]; let recent = &messages[split_point..]; + // TODO: before summarizing, check for an operator-provided + // `memory_flush` tool and invoke it with the to-compact slice + // so the operator can extract long-term facts. See plan §5. + let summary = summarize(llm, to_compact).await?; info!( "Compacted {} messages into summary ({} chars)", @@ -45,32 +74,31 @@ pub async fn compact_messages( summary.len() ); - let mut result = Vec::new(); + // Working context: system prompt + compaction summary + recent. + let mut working = Vec::new(); if let Some(sys) = system_msg { - result.push(sys); + working.push(sys); } - result.push(Message { - role: Role::User, + working.push(Message { + role: Role::System, content: Some(Content::Text(format!( - "[Conversation summary]\n{}", + "\n{}\n", summary ))), tool_calls: None, tool_call_id: None, }); - result.push(Message { - role: Role::Assistant, - content: Some(Content::Text( - "Understood. I have the conversation context from the summary. \ - Continuing from where we left off." - .to_string(), - )), - tool_calls: None, - tool_call_id: None, - }); - result.extend_from_slice(recent); + working.extend_from_slice(recent); + + // Canonical messages: only the real recent messages, no + // scaffolding. This is what the operator should persist. + let canonical = recent.to_vec(); - Ok(result) + Ok(CompactionResult { + working_messages: working, + canonical_messages: canonical, + summary, + }) } /// Walk the proposed split backward until no tool_call in diff --git a/loop/src/config.rs b/loop/src/config.rs index f956e3e..215946e 100644 --- a/loop/src/config.rs +++ b/loop/src/config.rs @@ -10,6 +10,16 @@ pub struct Config { pub mcp_servers: Vec, pub max_iterations: usize, pub timeout_minutes: u64, + /// Total context window in tokens. Env: `CONTEXT_WINDOW`. + pub context_window: usize, + /// Fraction of the context window at which auto-compaction kicks + /// in. Env: `COMPACTION_THRESHOLD`. + pub compaction_threshold: f64, + /// Number of most-recent messages kept during compaction. Env: + /// `COMPACTION_KEEP_RECENT`. + pub compaction_keep_recent: usize, + /// Max compaction rounds per session. Env: `MAX_COMPACTIONS`. + pub max_compactions: usize, /// Process-local identity. Attached as a tracing span field and, /// when the `sentry` feature is enabled, as a Sentry tag. pub agent_name: Option, @@ -59,6 +69,26 @@ impl Config { .parse() .unwrap_or(60); + let context_window: usize = env::var("CONTEXT_WINDOW") + .unwrap_or_else(|_| "128000".into()) + .parse() + .unwrap_or(128_000); + + let compaction_threshold: f64 = env::var("COMPACTION_THRESHOLD") + .unwrap_or_else(|_| "0.80".into()) + .parse() + .unwrap_or(0.80); + + let compaction_keep_recent: usize = env::var("COMPACTION_KEEP_RECENT") + .unwrap_or_else(|_| "10".into()) + .parse() + .unwrap_or(10); + + let max_compactions: usize = env::var("MAX_COMPACTIONS") + .unwrap_or_else(|_| "3".into()) + .parse() + .unwrap_or(3); + let agent_name = env::var("OVERLOOP_AGENT_NAME") .ok() .map(|s| s.trim().to_string()) @@ -105,6 +135,10 @@ impl Config { mcp_servers, max_iterations, timeout_minutes, + context_window, + compaction_threshold, + compaction_keep_recent, + max_compactions, agent_name, sentry_dsn, sentry_environment, @@ -133,6 +167,10 @@ mod tests { "MCP_SERVERS", "MAX_ITERATIONS", "TIMEOUT_MINUTES", + "CONTEXT_WINDOW", + "COMPACTION_THRESHOLD", + "COMPACTION_KEEP_RECENT", + "MAX_COMPACTIONS", "OVERLOOP_AGENT_NAME", "SENTRY_DSN", "SENTRY_ENVIRONMENT", @@ -169,6 +207,10 @@ mod tests { assert!(cfg.mcp_servers.is_empty()); assert_eq!(cfg.max_iterations, 100); assert_eq!(cfg.timeout_minutes, 60); + assert_eq!(cfg.context_window, 128_000); + assert!((cfg.compaction_threshold - 0.80).abs() < f64::EPSILON); + assert_eq!(cfg.compaction_keep_recent, 10); + assert_eq!(cfg.max_compactions, 3); assert!(cfg.agent_name.is_none()); assert!(cfg.sentry_dsn.is_none()); assert_eq!(cfg.sentry_environment, "local"); diff --git a/loop/src/main.rs b/loop/src/main.rs index a90d93a..4e6bfba 100644 --- a/loop/src/main.rs +++ b/loop/src/main.rs @@ -177,6 +177,10 @@ async fn run(config: Config) -> Result<()> { max_iterations: config.max_iterations, timeout: Duration::from_secs(config.timeout_minutes * 60), model: config.model.clone(), + context_window: config.context_window, + compaction_threshold: config.compaction_threshold, + compaction_keep_recent: config.compaction_keep_recent, + max_compactions: config.max_compactions, langfuse_capture_input: config.langfuse_capture_input, }; diff --git a/loop/src/observability/langfuse.rs b/loop/src/observability/langfuse.rs index af20707..3090a97 100644 --- a/loop/src/observability/langfuse.rs +++ b/loop/src/observability/langfuse.rs @@ -330,6 +330,10 @@ mod tests { mcp_servers: vec![], max_iterations: 1, timeout_minutes: 1, + context_window: 128_000, + compaction_threshold: 0.80, + compaction_keep_recent: 10, + max_compactions: 3, agent_name: None, sentry_dsn: None, sentry_environment: "local".into(), diff --git a/loop/src/traits.rs b/loop/src/traits.rs index 791f2ec..df86ab3 100644 --- a/loop/src/traits.rs +++ b/loop/src/traits.rs @@ -51,10 +51,23 @@ pub trait AcpService { /// returns. `id` echoes the one from `stream_tool_call`. fn stream_tool_result(&mut self, id: &str, content: &Value, is_error: bool) -> Result<()>; - /// Fire-and-forget notification emitted when a turn completes, - /// carrying the turn's messages and usage totals. The broker fans - /// this out to SSE subscribers. + /// Fire-and-forget notification emitted when a turn completes. + /// The broker fans this out to SSE subscribers. + /// + /// **`messages` is deprecated** — agents should send only usage. + /// The implementation sends an empty `messages` array. fn turn_end(&mut self, messages: &[Message], usage: &Usage) -> Result<()>; + + /// Fire-and-forget notification emitted after the agent compacts + /// its working context. Carries the prose summary of dropped + /// messages and the surviving canonical messages for operator + /// persistence. + fn context_compacted( + &mut self, + summary: &str, + messages: &[Message], + usage: &Usage, + ) -> Result<()>; fn quota_check(&mut self) -> Result; fn quota_update(&mut self, input_tokens: u64, output_tokens: u64) -> Result<()>; /// Block until the next `session/message` or `session/cancel` diff --git a/loop/tests/test_agentic_loop.rs b/loop/tests/test_agentic_loop.rs index db2c76c..e274fa8 100644 --- a/loop/tests/test_agentic_loop.rs +++ b/loop/tests/test_agentic_loop.rs @@ -136,6 +136,15 @@ impl AcpService for MockAcp { Ok(()) } + fn context_compacted( + &mut self, + _summary: &str, + _messages: &[Message], + _usage: &Usage, + ) -> Result<()> { + Ok(()) + } + fn quota_check(&mut self) -> Result { Ok(self.quota_allowed) } @@ -213,6 +222,10 @@ fn default_config() -> LoopConfig { max_iterations: 10, timeout: Duration::from_secs(30), model: "test-model".into(), + context_window: 128_000, + compaction_threshold: 0.80, + compaction_keep_recent: 10, + max_compactions: 3, langfuse_capture_input: false, } } @@ -425,6 +438,10 @@ async fn test_timeout() { max_iterations: 10, timeout: Duration::from_millis(0), model: "test-model".into(), + context_window: 128_000, + compaction_threshold: 0.80, + compaction_keep_recent: 10, + max_compactions: 3, langfuse_capture_input: false, }; @@ -543,6 +560,10 @@ async fn test_wind_down_injects_system_message_when_iterations_remaining_equals_ max_iterations: 5, timeout: Duration::from_secs(30), model: "test-model".into(), + context_window: 128_000, + compaction_threshold: 0.80, + compaction_keep_recent: 10, + max_compactions: 3, langfuse_capture_input: false, }; diff --git a/loop/tests/test_compaction.rs b/loop/tests/test_compaction.rs index e7412a6..1244828 100644 --- a/loop/tests/test_compaction.rs +++ b/loop/tests/test_compaction.rs @@ -74,7 +74,9 @@ async fn test_short_no_compaction() { let result = compact_messages(&llm, &messages, 4).await.unwrap(); // 3 messages <= keep_recent(4) + 2, so no compaction - assert_eq!(result.len(), 3); + assert_eq!(result.working_messages.len(), 3); + assert_eq!(result.canonical_messages.len(), 3); + assert!(result.summary.is_empty()); } #[tokio::test] @@ -89,10 +91,15 @@ async fn test_preserves_system_prompt() { let result = compact_messages(&llm, &messages, 5).await.unwrap(); - // First message should be the system prompt - assert_eq!(result[0].role, Role::System); + // First working message should be the system prompt + assert_eq!(result.working_messages[0].role, Role::System); assert_eq!( - result[0].content.as_ref().unwrap().as_text().unwrap(), + result.working_messages[0] + .content + .as_ref() + .unwrap() + .as_text() + .unwrap(), "You are helpful." ); } @@ -108,25 +115,26 @@ async fn test_compact_structure() { let result = compact_messages(&llm, &messages, 5).await.unwrap(); - // Structure: [system, summary_user, ack_assistant, ...5 recent] - assert_eq!(result.len(), 8); - assert_eq!(result[0].role, Role::System); - assert_eq!(result[1].role, Role::User); - assert!(result[1] + // Working: [system, system, ...5 recent] = 7 + assert_eq!(result.working_messages.len(), 7); + assert_eq!(result.working_messages[0].role, Role::System); + assert_eq!(result.working_messages[1].role, Role::System); + assert!(result.working_messages[1] .content .as_ref() .unwrap() .as_text() .unwrap() - .contains("[Conversation summary]")); - assert_eq!(result[2].role, Role::Assistant); - assert!(result[2] - .content - .as_ref() - .unwrap() - .as_text() - .unwrap() - .contains("Understood")); + .contains("")); + + // Canonical: only the 5 recent messages, no scaffolding + assert_eq!(result.canonical_messages.len(), 5); + for msg in &result.canonical_messages { + assert_ne!(msg.role, Role::System); + } + + // Summary should be populated + assert_eq!(result.summary, "Summary of earlier conversation."); } #[tokio::test] @@ -139,15 +147,42 @@ async fn test_no_system_prompt() { let result = compact_messages(&llm, &messages, 5).await.unwrap(); - // No system prompt in result - assert_ne!(result[0].role, Role::System); - // First should be the summary user message - assert_eq!(result[0].role, Role::User); - assert!(result[0] + // Working: [ system, ...5 recent] = 6 + assert_eq!(result.working_messages.len(), 6); + assert_eq!(result.working_messages[0].role, Role::System); + assert!(result.working_messages[0] .content .as_ref() .unwrap() .as_text() .unwrap() - .contains("[Conversation summary]")); + .contains("")); + + // Canonical: 5 recent, no system messages + assert_eq!(result.canonical_messages.len(), 5); + for msg in &result.canonical_messages { + assert_ne!(msg.role, Role::System); + } +} + +#[tokio::test] +async fn test_canonical_excludes_compaction_scaffolding() { + let llm = MockLlm { + summary: "Test summary".to_string(), + }; + + let mut messages = vec![make_msg(Role::System, "System prompt.")]; + messages.extend(make_conversation(20)); + + let result = compact_messages(&llm, &messages, 5).await.unwrap(); + + // No canonical message should contain + for msg in &result.canonical_messages { + if let Some(text) = msg.content.as_ref().and_then(|c| c.as_text()) { + assert!( + !text.contains(""), + "canonical messages must not contain compaction scaffolding" + ); + } + } } diff --git a/loop/tests/test_langfuse.rs b/loop/tests/test_langfuse.rs index 78e56d2..92c741e 100644 --- a/loop/tests/test_langfuse.rs +++ b/loop/tests/test_langfuse.rs @@ -23,6 +23,10 @@ fn test_config(host: &str, public: Option<&str>, secret: Option<&str>) -> Config mcp_servers: vec![], max_iterations: 1, timeout_minutes: 1, + context_window: 128_000, + compaction_threshold: 0.80, + compaction_keep_recent: 10, + max_compactions: 3, agent_name: None, sentry_dsn: None, sentry_environment: "local".into(), diff --git a/protocol/src/lib.rs b/protocol/src/lib.rs index ac95954..39227ff 100644 --- a/protocol/src/lib.rs +++ b/protocol/src/lib.rs @@ -24,8 +24,8 @@ pub use jwt::{ ROLE_AGENT, }; pub use messages::{ - Activity, Content, Heartbeat, InitializeRequest, InitializeResponse, Message, - QuotaCheckRequest, QuotaCheckResponse, QuotaUpdateRequest, QuotaUpdateResponse, Role, + Activity, Content, ContextCompactedParams, Heartbeat, InitializeRequest, InitializeResponse, + Message, QuotaCheckRequest, QuotaCheckResponse, QuotaUpdateRequest, QuotaUpdateResponse, Role, SessionCancelParams, SessionMessageParams, TextDelta, ToolCallNotification, ToolResultNotification, TurnEndParams, Usage, }; diff --git a/protocol/src/messages.rs b/protocol/src/messages.rs index 9f36776..a08210c 100644 --- a/protocol/src/messages.rs +++ b/protocol/src/messages.rs @@ -142,12 +142,35 @@ pub struct QuotaUpdateResponse {} // ── turn/end (agent → server, fire-and-forget notification) ───────── /// Params of the `turn/end` notification emitted by the agent when -/// a turn completes. The broker fans this out to SSE subscribers; -/// the operator's backend is responsible for persisting the data. -/// There is no response — this is a notification, not a request. +/// a turn completes. The broker fans this out to SSE subscribers. +/// +/// **`messages` is deprecated.** Agents SHOULD omit this field (it +/// will serialize as absent when the vec is empty). Operators that +/// need per-turn persistence should reconstruct the turn from +/// `stream/*` notifications, or wait for `context/compacted` which +/// carries the authoritative post-compaction history. #[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] pub struct TurnEndParams { + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub messages: Vec, + #[serde(default)] + pub usage: Usage, +} + +/// Params of the `context/compacted` notification. Emitted by the +/// agent after it compacts its working context. Carries a prose +/// summary of the messages that were dropped and the surviving +/// recent messages (canonical — no agent-internal scaffolding). +/// +/// The operator SHOULD replace its stored history for this +/// conversation with `messages` and record `summary` as the +/// compaction prefix so that a future `BootProvider::initialize` +/// can return both. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ContextCompactedParams { + /// LLM-generated summary of the messages that were dropped. + pub summary: String, + /// The messages that survived compaction (most recent window). pub messages: Vec, #[serde(default)] pub usage: Usage, diff --git a/protocol/src/methods.rs b/protocol/src/methods.rs index b924c19..e77e629 100644 --- a/protocol/src/methods.rs +++ b/protocol/src/methods.rs @@ -48,11 +48,20 @@ pub const TOOLS_CALL: &str = "tools/call"; // ── Turn lifecycle + quota ────────────────────────────────────────── /// Agent → server. Fire-and-forget notification marking the end of -/// a turn; carries the turn's messages and usage. The broker fans it -/// out to SSE subscribers; the operator's backend is responsible for -/// persisting the data. +/// a turn. The broker fans it out to SSE subscribers. +/// +/// **`messages` is deprecated** — agents SHOULD send only `usage`. +/// Per-turn persistence is reconstructed from `stream/*` +/// notifications; the authoritative post-compaction snapshot is +/// emitted via [`CONTEXT_COMPACTED`]. pub const TURN_END: &str = "turn/end"; +/// Agent → server. Fire-and-forget notification emitted after the +/// agent compacts its working context. Carries a prose `summary` of +/// the dropped messages and the surviving `messages` window. The +/// operator SHOULD persist this as the new canonical history. +pub const CONTEXT_COMPACTED: &str = "context/compacted"; + /// Agent → server. Ask the operator's `QuotaPolicy` whether the next /// turn is allowed. pub const QUOTA_CHECK: &str = "quota/check"; diff --git a/protocol/tests/fixtures/context_compacted.json b/protocol/tests/fixtures/context_compacted.json new file mode 100644 index 0000000..b75b7ce --- /dev/null +++ b/protocol/tests/fixtures/context_compacted.json @@ -0,0 +1,8 @@ +{ + "summary": "User asked to refactor auth middleware. We identified the legacy session token storage and agreed on JWT-based replacement.", + "messages": [ + { "role": "user", "content": "now deploy it" }, + { "role": "assistant", "content": "Deploying to staging..." } + ], + "usage": { "input_tokens": 1200, "output_tokens": 400 } +} diff --git a/protocol/tests/fixtures/turn_end_usage_only.json b/protocol/tests/fixtures/turn_end_usage_only.json new file mode 100644 index 0000000..9cf902e --- /dev/null +++ b/protocol/tests/fixtures/turn_end_usage_only.json @@ -0,0 +1,3 @@ +{ + "usage": { "input_tokens": 5000, "output_tokens": 1200 } +} diff --git a/protocol/tests/wire_fixtures.rs b/protocol/tests/wire_fixtures.rs index cedc8c0..09b4bb0 100644 --- a/protocol/tests/wire_fixtures.rs +++ b/protocol/tests/wire_fixtures.rs @@ -3,8 +3,8 @@ //! values are equal (key order ignored). use overacp_protocol::messages::{ - Activity, Heartbeat, InitializeResponse, Message, QuotaCheckResponse, QuotaUpdateRequest, - SessionCancelParams, SessionMessageParams, TextDelta, ToolCallNotification, + Activity, ContextCompactedParams, Heartbeat, InitializeResponse, Message, QuotaCheckResponse, + QuotaUpdateRequest, SessionCancelParams, SessionMessageParams, TextDelta, ToolCallNotification, ToolResultNotification, TurnEndParams, }; use serde::de::DeserializeOwned; @@ -41,6 +41,16 @@ fn turn_end_notification_roundtrip() { assert_roundtrip::(include_str!("fixtures/turn_end_notification.json")); } +#[test] +fn turn_end_usage_only_roundtrip() { + assert_roundtrip::(include_str!("fixtures/turn_end_usage_only.json")); +} + +#[test] +fn context_compacted_roundtrip() { + assert_roundtrip::(include_str!("fixtures/context_compacted.json")); +} + #[test] fn session_message_roundtrip() { assert_roundtrip::(include_str!("fixtures/session_message.json")); diff --git a/server/src/tunnel/dispatch.rs b/server/src/tunnel/dispatch.rs index 54686ed..ae47bce 100644 --- a/server/src/tunnel/dispatch.rs +++ b/server/src/tunnel/dispatch.rs @@ -57,10 +57,10 @@ pub async fn handle_message(text: &str, ctx: &TunnelContext) -> Option { )), // Notifications — agent → server, no response. The read loop - // fans `stream/*`, `turn/end`, and `heartbeat` out to the - // in-memory broker for SSE subscribers. + // fans `stream/*`, `turn/end`, `context/compacted`, and + // `heartbeat` out to the in-memory broker for SSE subscribers. "stream/textDelta" | "stream/activity" | "stream/toolCall" | "stream/toolResult" - | "turn/end" | "heartbeat" => { + | "turn/end" | "context/compacted" | "heartbeat" => { debug!(method, %agent_id, "tunnel notification"); None } @@ -306,13 +306,24 @@ mod tests { async fn turn_end_is_a_notification() { let ctx = ctx_default(Uuid::new_v4()); let resp = handle_message( - r#"{"jsonrpc":"2.0","method":"turn/end","params":{"messages":[]}}"#, + r#"{"jsonrpc":"2.0","method":"turn/end","params":{"usage":{"input_tokens":100,"output_tokens":50}}}"#, &ctx, ) .await; assert!(resp.is_none(), "turn/end is fire-and-forget"); } + #[tokio::test] + async fn context_compacted_is_a_notification() { + let ctx = ctx_default(Uuid::new_v4()); + let resp = handle_message( + r#"{"jsonrpc":"2.0","method":"context/compacted","params":{"summary":"test","messages":[],"usage":{"input_tokens":0,"output_tokens":0}}}"#, + &ctx, + ) + .await; + assert!(resp.is_none(), "context/compacted is fire-and-forget"); + } + #[tokio::test] async fn session_cancel_from_agent_is_method_not_found() { // session/cancel is server → agent only. If the agent ever diff --git a/server/src/tunnel/run.rs b/server/src/tunnel/run.rs index 7b0c02e..6661d09 100644 --- a/server/src/tunnel/run.rs +++ b/server/src/tunnel/run.rs @@ -130,12 +130,13 @@ pub async fn run_tunnel(ws: WebSocket, claims: Claims, ctx: Arc) trace!(%agent_id, payload = %text, "agent→server"); ctx.registry.touch(agent_id).await; - // Best-effort fan-out of stream/* and turn/end - // notifications to the broker so SSE subscribers - // receive them. Cheap string sniff to avoid - // parsing every frame twice. + // Best-effort fan-out of stream/*, turn/end, + // and context/compacted notifications to the + // broker so SSE subscribers receive them. Cheap + // string sniff to avoid parsing every frame twice. if text.contains("\"stream/") || text.contains("\"turn/end\"") + || text.contains("\"context/compacted\"") || text.contains("\"heartbeat\"") { ctx.stream_broker