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