feat(protocol,loop): agent-owned context, operator-owned record

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

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<ToolCall>` 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<ToolCall>` 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. |

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
+      `<compacted_context>` 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 `<turn>` 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

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<serde_json::Value>,
     _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<S>(stream: &mut S) -> Result<Option<serde_json::Value>, 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<S>(
+    stream: &mut S,
+) -> Result<Option<(serde_json::Value, Vec<serde_json::Value>)>, reqwest::Error>
 where
     S: Stream<Item = Result<Bytes, reqwest::Error>> + 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::<serde_json::Value>(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",
+    );
 }

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,