From 6ca7b151f715caafc49bda394aa329b9267a93e9 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 5 Nov 2026 22:07:12 -0600 Subject: [PATCH] feat: compact an agent's context from the task Properties panel (#19378) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A task's Properties panel shows the assignee agent's context window (paperclipai/paperclip#19273) > - The board can see that the context is nearly full, but it cannot act on it > - A full context makes the agent compact at an unplanned time, often in the middle of work > - This pull request adds "Compact now" to the Context hover card > - The benefit is that the board can compact the agent's context between runs, at a time it chooses ## Linked Issues or Issue Description Refs paperclipai/paperclip#19273 **Problem** The Context row shows how full the agent's context window is. The board cannot ask the agent to compact it. **Proposed solution** A "Compact now" button in the Context hover card. It runs the harness's own compaction: Codex `thread/compact/start`, and Claude's `/compact` command through ACP. **Alternatives considered** A separate PRP `session.compact` command. Both harnesses run compaction as a turn, so a compaction turn keeps the existing durable turn binding, interruption, and recovery. A separate command needs a second turn lifecycle. ## What Changed - UI: the Context hover card shows Compact now with three states: available, Compacting…, and unavailable with a reason (agent mid-run, harness unsupported, task closed). The Storybook story shows each state. - Server: `POST /api/issues/:id/context/compact` (board only, `decideIssueAccess` `issue:mutate`, activity log `issue.context_compaction_requested`). It queues a short run marked by `contextSnapshot.contextCompactionRequestId`. `GET /issues/:id/context-usage` returns the `compaction` state. - Server: a compaction run does no task work. The native status arbiter preserves the task status. The lifecycle skips handoff, liveness continuation, review-path disposition and the issue comment, as it does for goal-control runs. - Runner TS: `startTurn({ compact: true })`. The Codex harness session sends `thread/compact/start`. The runnerd transport maps it to PRP `turn.start` with `compact: true` (Codex) or `text: "/compact"` (Claude/ACPX). The native runtime option `sessionCompaction` requires a resumed session and settles with a synthesized yielded result. - Runner Rust: `turn.start` accepts `compact: true` for Codex. `CodexProvider::start_compaction_turn` sends `thread/compact/start` and binds the turn from the `turn/started` that follows the empty response. - Fix (from #19273): Codex emitted `context.compacted` on both the start and the end of one compaction item. It now emits it once, and the count is distinct by `compactionId`. - Fix (from #19273): the conversation event view dropped `context.compacted`, so a finished run never showed the divider. The view now keeps compactions, plus context readings for runs that compacted. A finished run's compactions show as thread dividers. - Fix: the Codex harness stamps before/after sizes on the compaction event. Codex reports the compacted size before the compaction item ends, so the old neighbour-based guess showed the wrong size. ## Verification - `pnpm -r typecheck`, `pnpm build`, `pnpm check:token-gates`, `node scripts/storybook-catalog.mjs`: pass. - New tests: `server/src/__tests__/context-compaction.test.ts`, `status-arbiter.test.ts`, `issue-context-usage.test.ts` (duplicate compaction counts once), `ui/.../context-usage-row.test.tsx`, `native-session-runtime.test.ts` (context compaction), Codex driver and runnerd transport tests, Rust `codex_compaction_turn_binds_the_turn_reported_after_its_empty_response`. Rust `codex_provider` suite: 115 passed. - `pnpm test:run` locally: one failure in `chat-channels.integration.test.ts`. It reads `.claude/skills/slack/SKILL.md` from the main checkout, which is not present in this worktree. It is not related to this change. - Live, local worktree instance, Codex agent on the Paperclip runner: clicked Compact now. The run succeeded. Context went from 35k to 9.4k tokens, and later 9.4k to 9.2k. The task stayed In Review. The thread shows "Context compacted · 9.4k → 9.2k tokens" and the meter dropped. - Live Claude proof: pending. The local Claude agent needs an Anthropic API key. claude-agent-acp handles `/compact` as an echo-less turn, and the runner already maps its "Compacting completed." notice to `context.compacted`. ## Risks - A compaction run resumes the agent's session. If the session cannot resume, the run fails with `native_context_compaction_unavailable` and the task is unchanged. - Compact now is not available while the agent runs on the task, because compaction interrupts the active turn. - The conversation view now returns context readings for runs that compacted. Runs that never compacted return the same events as before. - An old runner binary ignores `compact` and rejects the `turn.start` because it has no text. It does not start an ordinary turn. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, through Claude Code with tool use and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- .../src/bin/fake-codex-app-server.rs | 23 ++ .../crates/runner-core/src/codex_provider.rs | 127 ++++++++++ .../runner-core/src/provider_backend.rs | 64 +++-- .../crates/runner-core/src/provider_events.rs | 6 +- .../runner-core/tests/codex_provider.rs | 44 ++++ .../src/contracts/harness-driver.ts | 7 + .../src/contracts/native-session-backend.ts | 7 + ...-app-server-driver.context-control.test.ts | 12 + .../codex-app-server-driver.test-support.ts | 4 + .../drivers/codex/codex-harness-session.ts | 26 +- .../codex/codex-session-notifications.ts | 28 ++- .../src/drivers/codex/codex-session-state.ts | 8 + .../src/live/runnerd-codex-transport.test.ts | 9 + .../src/live/runnerd-codex-transport.ts | 19 +- .../src/native-session-runtime.test.ts | 225 ++++++++++++++++++ .../src/native-session-runtime.ts | 87 ++++++- .../paperclip-runner/src/provider-events.ts | 5 +- packages/shared/src/index.ts | 2 +- packages/shared/src/issue-context-usage.ts | 12 + .../src/__tests__/context-compaction.test.ts | 106 +++++++++ .../__tests__/heartbeat-auto-checkout.test.ts | 1 + .../src/__tests__/issue-context-usage.test.ts | 3 + .../modules/wake-queue/domain/context.test.ts | 11 + .../src/modules/wake-queue/domain/context.ts | 4 + server/src/routes/activity.ts | 6 +- server/src/routes/issues.ts | 59 +++++ server/src/routes/openapi.ts | 10 + server/src/services/context-compaction.ts | 201 ++++++++++++++++ server/src/services/heartbeat.ts | 7 + .../services/heartbeat/conversation-events.ts | 20 ++ .../src/services/heartbeat/run-completion.ts | 3 + .../src/services/heartbeat/run-lifecycle.ts | 14 +- .../services/heartbeat/runtime-execution.ts | 2 + server/src/services/issue-context-usage.ts | 7 +- .../native-runtime/native-run-finalizer.ts | 2 + .../native-runtime/native-session-executor.ts | 3 + .../native-runtime/status-arbiter.test.ts | 12 + .../services/native-runtime/status-arbiter.ts | 12 + server/src/services/recovery/service.ts | 2 + ui/src/api/issues.ts | 2 + ui/src/components/ContextUsageMeter.tsx | 49 +++- ui/src/components/IssueProperties.test.tsx | 1 + ui/src/components/TaskChatThread.tsx | 10 +- .../issue-properties/IssueProperties.tsx | 2 +- .../context-usage-row.test.tsx | 56 ++++- .../issue-properties/context-usage-row.tsx | 55 ++++- .../task-chat/transcript-adapter.ts | 21 +- .../tasks/task-context-window.stories.tsx | 101 ++++++-- 48 files changed, 1431 insertions(+), 66 deletions(-) create mode 100644 server/src/__tests__/context-compaction.test.ts create mode 100644 server/src/services/context-compaction.ts diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs index ed44bfc42ee..b56ffc364bc 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs @@ -1596,6 +1596,29 @@ fn run() -> Result<(), Box> { } } } + "thread/compact/start" => { + // Codex answers before the compaction turn starts and binds + // its identity only through turn/started. + send(json!({"id": id, "result": {}}))?; + let turn_id = "provider-compact-turn-1"; + send(json!({"method": "turn/started", "params": { + "threadId": state.thread_id, "turn": {"id": turn_id, "status": "inProgress"} + }}))?; + let item = json!({"id": "compact-item-1", "type": "contextCompaction"}); + send(json!({"method": "item/started", "params": { + "threadId": state.thread_id, "turnId": turn_id, "item": item + }}))?; + send(json!({"method": "thread/compacted", "params": { + "threadId": state.thread_id, "turnId": turn_id + }}))?; + send(json!({"method": "item/completed", "params": { + "threadId": state.thread_id, "turnId": turn_id, "item": item + }}))?; + send(json!({"method": "turn/completed", "params": { + "threadId": state.thread_id, + "turn": {"id": turn_id, "status": "completed", "items": []} + }}))?; + } "thread/goal/clear" => { state.goal = None; save_state(&state_path, &state)?; diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs index abfc3bdb241..e4eb31abf5e 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs @@ -2235,6 +2235,133 @@ impl CodexProvider { ) } + /// Starts Codex's on-demand context compaction. Codex runs compaction as + /// its own turn, but `thread/compact/start` answers `{}`, so the turn + /// identity is bound from the `turn/started` notification that follows. + /// Messages read while waiting stay buffered for normal event delivery. + pub fn start_compaction_turn(&mut self) -> Result { + if self.config.provider != "codex" { + return Err(LocalRunnerError::invalid( + "on-demand context compaction requires Codex", + )); + } + if self.quarantined { + return Err(LocalRunnerError::invalid( + "Codex provider is quarantined after unsafe recovered work", + )); + } + if self.active_provider_turn_id.is_some() { + return Err(LocalRunnerError::invalid( + "Codex already has an active provider turn", + )); + } + if self.ambiguous_turn_start_pending { + return Err(LocalRunnerError::invalid( + "Codex has an unresolved ambiguous provider turn start", + )); + } + self.rollover_settled_turn_epoch_if_needed()?; + let prior_reconciliation_pending = self.completion_reconciliation_pending; + self.completion_reconciliation_pending = false; + let prior_buffered_message_count = self.pending_messages.len(); + self.ambiguous_turn_start_pending = true; + let runtime_request_scope = new_runtime_request_scope()?; + match self.request_classified("thread/compact/start", json!({"threadId": self.thread_id})) { + Ok(_) => {} + Err(ProviderRequestError::Rejected(error)) => { + let definite_rejection = self + .pending_messages + .iter() + .skip(prior_buffered_message_count) + .all(|buffered| is_unbound_rejected_turn_diagnostic(&buffered.value)); + if definite_rejection { + self.ambiguous_turn_start_pending = false; + self.completion_reconciliation_pending = prior_reconciliation_pending; + } + return Err(error); + } + Err(ProviderRequestError::Ambiguous(error)) => return Err(error), + } + let deadline = std::time::Instant::now() + Duration::from_secs(30); + let mut scanned = prior_buffered_message_count; + loop { + while scanned < self.pending_messages.len() { + let message = &self.pending_messages[scanned].value; + scanned += 1; + if message.get("method").and_then(Value::as_str) != Some("turn/started") + || message.get("id").is_some() + { + continue; + } + let params = message.get("params").cloned().unwrap_or(Value::Null); + validate_notification_binding(&self.thread_id, None, ¶ms)?; + let provider_turn_id = match bounded_provider_turn_id(notification_turn_id(¶ms)) + { + Ok(provider_turn_id) => provider_turn_id, + Err(error) => { + // Codex accepted the compaction without a usable + // identity. It cannot be bound or interrupted, so + // terminate it like an invalid turn/start response. + self.rejected_accepted_turn = Some(RejectedAcceptedTurn::InvalidIdentity); + self.expected_shutdown = true; + self.completed_turn_authority = None; + let _ = self.cancel_pending_requests(); + let _ = self.process.terminate_group(); + return Err(error); + } + }; + if self.settled_provider_turn_ids.contains(&provider_turn_id) { + return Err(self.reject_accepted_reused_turn_identity(provider_turn_id)); + } + self.accept_replacement_turn(provider_turn_id.clone(), runtime_request_scope); + return Ok(json!({"turn": {"id": provider_turn_id}})); + } + let remaining = deadline.saturating_duration_since(std::time::Instant::now()); + if remaining.is_zero() { + // The start stays ambiguous: a late turn/started is still + // adopted by the ambiguous-start validator. + return Err(LocalRunnerError::invalid( + "Codex thread/compact/start did not report its turn within 30 seconds", + )); + } + let Some(line) = self.receive_provider_stdout_line(remaining)? else { + if let Some(exit) = self.process.try_wait()? { + self.drain_provider_diagnostics(Duration::from_millis(50)); + let diagnostic_suffix = self.provider_diagnostic_suffix(); + return Err(LocalRunnerError::invalid(format!( + "Codex exited before its compaction turn started (exitCode={:?}, signal={:?}){diagnostic_suffix}", + exit.exit_code, exit.signal + ))); + } + continue; + }; + let trace_frame_id = self.trace_inbound(&line); + let message = parse_provider_message(&line)?; + if self + .pending_messages + .len() + .saturating_add(self.deferred_ambiguous_messages.len()) + >= MAX_BUFFERED_MESSAGES + { + return Err(LocalRunnerError::invalid( + "Codex emitted too many messages before its compaction turn started", + )); + } + let retained_bytes = json_size(&message, "buffered Codex provider message")?; + self.pending_message_bytes = + retain_buffered_message_bytes(self.pending_message_bytes, retained_bytes) + .ok_or_else(|| { + LocalRunnerError::invalid( + "Codex buffered messages exceed the 16 MiB aggregate limit", + ) + })?; + self.pending_messages.push_back(BufferedProviderMessage { + value: message, + trace_frame_id, + }); + } + } + pub fn steer_turn(&mut self, message: &str) -> Result { let turn_id = self .active_provider_turn_id diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs index aa0101022f8..2588c4d049c 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs @@ -3619,6 +3619,24 @@ impl CodexCommandExecutor { } Some(_) => return Err(DurableRunnerError::invalid("invalid turn.reasoningMode")), }; + // `compact` starts the provider's on-demand context compaction as + // this turn. It carries no user text, skills, or reasoning overrides. + let compact = match payload.get("compact") { + None | Some(Value::Bool(false)) => false, + Some(Value::Bool(true)) => true, + Some(_) => { + return Err(DurableRunnerError::invalid( + "turn.start compact must be a boolean", + )) + } + }; + if compact + && (!skills.is_empty() || reasoning_mode.is_some() || payload.get("text").is_some()) + { + return Err(DurableRunnerError::invalid( + "a compaction turn.start cannot carry text, skills, or a reasoning mode", + )); + } self.restore_provider_if_needed()?; if reasoning_mode.is_some() && self @@ -3677,10 +3695,24 @@ impl CodexCommandExecutor { )) })?; self.rollover_provider_identity_epochs_if_needed()?; - let text = payload - .get("text") - .and_then(Value::as_str) - .ok_or_else(|| DurableRunnerError::invalid("turn.start payload.text is required"))?; + let text = if compact { + "" + } else { + payload + .get("text") + .and_then(Value::as_str) + .ok_or_else(|| DurableRunnerError::invalid("turn.start payload.text is required"))? + }; + if compact + && self + .state + .as_ref() + .is_some_and(|state| state.config.provider != "codex") + { + return Err(DurableRunnerError::invalid( + "on-demand context compaction requires Codex", + )); + } let cwd = self .state .as_ref() @@ -3716,16 +3748,20 @@ impl CodexCommandExecutor { rejected_accepted_turn, ) = { let provider = self.ensure_provider()?; - let result = provider.start_turn_with_options_and_effort( - text, - &cwd, - CodexTurnOptions { - skills: &skills, - reasoning_mode, - }, - payload.get("effort"), - run_id.as_deref(), - ); + let result = if compact { + provider.start_compaction_turn() + } else { + provider.start_turn_with_options_and_effort( + text, + &cwd, + CodexTurnOptions { + skills: &skills, + reasoning_mode, + }, + payload.get("effort"), + run_id.as_deref(), + ) + }; ( result, provider.completed_turn_authority().is_some(), diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs index 8414293cc76..e6dc4b10937 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs @@ -1008,7 +1008,11 @@ pub fn normalize_codex_notification_in( EventPriority::P1, json!({ "schema": "paperclip.context.compacted.v1", - "compactionId": stable_id(string(params.get("threadId")), "codex-compaction"), + // One thread compacts many times; its turn names each compaction. + "compactionId": stable_id( + &format!("{}:{}", string(params.get("threadId")), string(params.get("turnId"))), + "codex-compaction", + ), "reason": "provider", "preTokens": Value::Null, "postTokens": Value::Null, diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs index ca1e43808ac..b23adf70231 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs @@ -1345,6 +1345,50 @@ fn codex_transport_buffers_notifications_while_waiting_for_responses() { fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); } +#[test] +fn codex_compaction_turn_binds_the_turn_reported_after_its_empty_response() { + let directory = temporary_directory("compaction-turn"); + let config = provider_config(&directory, &[]); + let mut provider = CodexProvider::start(&config, None).expect("start fake Codex provider"); + let started = provider + .start_compaction_turn() + .expect("start compaction turn"); + assert_eq!( + started.pointer("/turn/id"), + Some(&json!("provider-compact-turn-1")) + ); + assert_eq!( + provider.active_provider_turn_id(), + Some("provider-compact-turn-1") + ); + // A second start while the compaction turn runs is refused. + assert!(provider.start_compaction_turn().is_err()); + + let mut event_types = Vec::new(); + let deadline = event_deadline(); + while std::time::Instant::now() < deadline { + if let Some(CodexProviderEvent::Notification { method, params }) = + provider.poll().expect("poll provider event") + { + event_types.extend( + normalize_codex_notification(&method, ¶ms) + .into_iter() + .map(|event| event.event_type), + ); + } + if event_types.iter().any(|event| event == "turn.completed") { + break; + } + std::thread::sleep(std::time::Duration::from_millis(1)); + } + assert!(event_types.iter().any(|event| event == "turn.started")); + assert!(event_types.iter().any(|event| event == "context.compacted")); + assert!(event_types.iter().any(|event| event == "turn.completed")); + assert_eq!(provider.active_provider_turn_id(), None); + provider.shutdown().expect("stop provider"); + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + #[test] fn codex_account_updates_do_not_interrupt_turns_or_publish_account_details() { let directory = temporary_directory("account-notifications"); diff --git a/packages/paperclip-runner/src/contracts/harness-driver.ts b/packages/paperclip-runner/src/contracts/harness-driver.ts index 5914b4a6550..94086d85ec8 100644 --- a/packages/paperclip-runner/src/contracts/harness-driver.ts +++ b/packages/paperclip-runner/src/contracts/harness-driver.ts @@ -538,6 +538,13 @@ export interface HarnessSession { requestedCollaborationMode?: "default" | "plan"; /** OpenCode/OpenRouter only. Applies to this turn, never subsequent turns. */ reasoningMode?: "default" | "disabled"; + /** + * Run the provider's own context compaction as this turn instead of + * submitting `message`. Only for a resumed session with prior context. + */ + compact?: true; + /** Last context occupancy Paperclip recorded before a compaction turn. */ + contextTokensBefore?: number; }): Promise<{ turnId: string; effectiveCollaborationMode?: "default" | "plan"; diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index 07b829c0fd5..aefb325d705 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -205,6 +205,13 @@ export interface NativeSession { requestedCollaborationMode?: "default" | "plan"; /** OpenCode/OpenRouter only. Applies to this turn, never subsequent turns. */ reasoningMode?: "default" | "disabled"; + /** + * Run the provider's own context compaction as this turn instead of + * submitting `message`. Only for a resumed session with prior context. + */ + compact?: true; + /** Last context occupancy Paperclip recorded before a compaction turn. */ + contextTokensBefore?: number; }): Promise<{ turnId: string; effectiveCollaborationMode?: "default" | "plan"; diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.context-control.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.context-control.test.ts index 332522cb698..719c7035661 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.context-control.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.context-control.test.ts @@ -58,6 +58,18 @@ describe("Codex app-server Codex driver", () => { await session.close({ reason: "fixture complete" }); }); + it("starts a compaction turn through thread/compact/start without submitting a message", async () => { + const transport = new FakeCodexTransport(); + const session = await makeDriver([transport]).openSession({ + runId: "run-compact", normalizedSessionId: "normalized-compact", workingDirectory: WORKSPACE, + }); + const started = await session.startTurn({ message: { role: "user", text: "" }, continuation: true, compact: true }); + expect(started.turnId).toBe("turn-compact"); + expect(transport.calls.filter(call => call.method === "turn/start")).toHaveLength(0); + expect(transport.calls.find(call => call.method === "thread/compact/start")?.params).toEqual({ threadId: "thread-1" }); + await session.close({ reason: "fixture complete" }); + }); + it("retains one correlation-bound steering acknowledgement after the rehydrated runnerd echo", async () => { const transport = new FakeCodexTransport("thread-root"); const session = await makeDriver([transport]).openSession({ diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test-support.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test-support.ts index b2d9a09644b..a797fa946d7 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test-support.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test-support.ts @@ -182,6 +182,10 @@ export class FakeCodexTransport implements CodexAppServerTransport { } ); } + if (method === "thread/compact/start") { + // The runnerd transport binds the compaction turn like turn/start. + return { turn: { id: "turn-compact", status: "inProgress", items: [] } }; + } if (method === "thread/goal/get") return { goal: this.goalState }; if (method === "thread/goal/set") { this.goalState = { diff --git a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts index 16e11c5ec00..c9c2ef14354 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts @@ -183,11 +183,23 @@ export class CodexHarnessSession requestedCollaborationMode?: "default" | "plan"; /** OpenCode/OpenRouter only. Applies to this turn, never subsequent turns. */ reasoningMode?: "default" | "disabled"; + /** Run the provider's context compaction as this turn. */ + compact?: true; + /** Last context occupancy recorded before a compaction turn. */ + contextTokensBefore?: number; }): Promise<{ turnId: string; effectiveCollaborationMode: "default" | "plan"; }> { this.assertProtocolIntegrity(); + if (input.compact === true && (input.reasoningMode !== undefined || this.dispositionOnlyRecoveryAvailable)) { + throw this.unsupported("context compaction", "compaction cannot combine with reasoning or disposition recovery"); + } + // A compaction-only run resumes the session without a fresh usage report; + // seed the "before" size the divider shows from Paperclip's last reading. + if (input.compact === true && input.contextTokensBefore !== undefined) { + this.lastContextUsedTokens ??= input.contextTokensBefore; + } if (input.reasoningMode !== undefined && !this.supportsTurnReasoning()) { throw new Error("Per-turn reasoning is not supported by this provider"); } @@ -241,8 +253,8 @@ export class CodexHarnessSession } // The submitted text is part of the canonical record so a tracer can show // the operator's own message without keeping shadow state next to the - // reducer. - this.emit("turn.submitted", { + // reducer. A compaction turn submits no message. + if (input.compact !== true) this.emit("turn.submitted", { envelopeSchema: this.taskEnvelope.schema, text: input.message.text, ...(turnSkills.length ? { skillInputs: turnSkills } : {}), @@ -258,7 +270,11 @@ export class CodexHarnessSession let response: Record; const requestedMode = this.opened.context.collaborationMode; try { - response = await this.transport.request("turn/start", { + // Compaction runs as the provider's own turn. Codex's response carries + // no turn; the runnerd transport binds it and answers like turn/start. + response = input.compact === true + ? await this.transport.request("thread/compact/start", { threadId: this.opened.threadId }) + : await this.transport.request("turn/start", { threadId: this.opened.threadId, ...(input.reasoningMode === undefined ? {} : { reasoningMode: input.reasoningMode }), ...(this.reasoningEffort ? { effort: this.reasoningEffort } : {}), @@ -320,7 +336,9 @@ export class CodexHarnessSession } this.assertProtocolIntegrity(); const turn = record(response.turn); - const turnId = text(turn.id); + // A direct Codex app-server answers thread/compact/start with `{}`; its + // compaction turn is known only from a turn/started that already arrived. + const turnId = text(turn.id) || (input.compact === true ? (this.activeTurnId ?? "") : ""); if (turnId.length === 0) { // A start notification is only optimistic until the response validates. // Clear it before released semantic/terminal waiters can observe an diff --git a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts index b16d2c72cb0..190bb1ab5db 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts @@ -277,12 +277,18 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex state.activeTurnId === null || turnId === state.activeTurnId) ) { + if (notification.method === "item/started" && record(params.item).type === "contextCompaction") { + state.compactionContext = { preTokens: state.lastContextUsedTokens, postTokens: null }; + } for (const canonical of canonicalProviderEventsFromCodex( notification.method, params, { workingDirectory: state.workspaceRoots }, )) { - state.emit(canonical.eventType, canonical.payload, { + const payload = canonical.eventType === "context.compacted" + ? withCompactionSizes(state, canonical.payload) + : canonical.payload; + state.emit(canonical.eventType, payload, { turnId: turnId || undefined, itemId: canonical.itemId, }); @@ -631,6 +637,11 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex } if (notification.method === "thread/tokenUsage/updated") { state.usageSnapshot = boundedPayload(record(params.tokenUsage)); + const contextUsedTokens = record(record(params.tokenUsage).context).usedTokens; + if (typeof contextUsedTokens === "number" && Number.isSafeInteger(contextUsedTokens) && contextUsedTokens >= 0) { + if (state.compactionContext) state.compactionContext.postTokens = contextUsedTokens; + else state.lastContextUsedTokens = contextUsedTokens; + } if (state.driverKind === "codex_app_server") { const reported = record(record(params.tokenUsage).total); const validCount = (value: unknown) => typeof value === "number" && Number.isSafeInteger(value) && value >= 0; @@ -665,6 +676,21 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex // Resume can re-deliver committed history for a turn the checkpoint already // settled. Between turns the identity classifier sees no active turn, so the // replay would otherwise reach the live-turn guards and fail the session. +/** Fills a compaction's missing sizes from the readings around it. */ +function withCompactionSizes( + state: CodexSessionState, + payload: Record, +): Record { + const count = (value: unknown) => + typeof value === "number" && Number.isSafeInteger(value) && value >= 0 ? value : null; + const observed = state.compactionContext; + state.compactionContext = null; + const preTokens = count(payload.preTokens) ?? observed?.preTokens ?? null; + const postTokens = count(payload.postTokens) ?? observed?.postTokens ?? null; + if (postTokens !== null) state.lastContextUsedTokens = postTokens; + return { ...payload, preTokens, postTokens }; +} + function ignoreRestoredTerminalReplay( state: CodexSessionState, method: string, diff --git a/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts b/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts index a9ace1defee..28f305421b3 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts @@ -119,6 +119,14 @@ export class CodexSessionState { activeTurnId: string | null; usageSnapshot: Record | null = null; codexUsageBaseline: CodexUsageBaseline | null = null; + /** Latest provider-reported context occupancy, outside a compaction. */ + lastContextUsedTokens: number | null = null; + /** + * Context sizes around an in-progress provider compaction. Codex reports the + * compacted size while its compaction item is still open, so the divider's + * "before → after" is captured here rather than from neighbouring events. + */ + compactionContext: { preTokens: number | null; postTokens: number | null } | null = null; /** Latest account rate-limit snapshot, merged across sparse updates. */ rateLimits: ProviderRateLimitSnapshot | null = null; rateLimitNoticeSequence = 0; diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts index fd6c8da5ac2..1c2b35e6f7c 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -5876,6 +5876,15 @@ it("rejects Pi thinking settings on a different transport provider", () => { expect(() => createCapabilityRunnerdCodexTransport({ provider: "codex", piThinkingLevel: "low" })).toThrow(/only supported/); }); +describe("runnerd compaction turn payload", () => { + it("asks Codex for its own compaction and Claude for its /compact command", async () => { + const { runnerdCompactionTurnPayload } = await import("./runnerd-codex-transport.js"); + expect(runnerdCompactionTurnPayload("codex")).toEqual({ compact: true }); + expect(runnerdCompactionTurnPayload("acpx")).toEqual({ text: "/compact" }); + expect(() => runnerdCompactionTurnPayload("opencode")).toThrow("not supported"); + }); +}); + describe("cold process admission budget", () => { it("bounds Pi cold startup and recovery at 60 seconds", () => { expect(coldAdmissionTimeoutMs("acpx", "pi", true)).toBe(60_000); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index d9d3cb1fb27..551fef7991e 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -1646,6 +1646,20 @@ class NotificationQueue implements AsyncIterable { } } +/** Internal marker: a thread/compact/start mapped onto the turn.start path. */ +const COMPACTION_TURN = "paperclipCompactionTurn"; + +/** + * Compaction runs as a provider turn so it keeps the normal turn binding. + * Codex compacts through app-server `thread/compact/start`; Claude's ACP agent + * compacts when prompted with its `/compact` command. + */ +export function runnerdCompactionTurnPayload(provider: string): Record { + if (provider === "codex") return { compact: true }; + if (provider === "acpx") return { text: "/compact" }; + throw new Error(`Context compaction is not supported by the ${provider} provider`); +} + function record(value: unknown): Record { return typeof value === "object" && value !== null && !Array.isArray(value) ? (value as Record) @@ -3900,6 +3914,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { : { data: [] }; } if (method === "turn/start") return this.#startTurn(params); + if (method === "thread/compact/start") return this.#startTurn({ threadId: params.threadId, [COMPACTION_TURN]: true }); if (method === "turn/steer") { const input = Array.isArray(params.input) ? params.input.map(record) : []; const text = input @@ -6627,6 +6642,8 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { throw new Error("turnStartTimeoutMs must be a positive safe integer"); } const commandDeadline = Date.now() + turnStartTimeoutMs; + const compactionTurn = params[COMPACTION_TURN] === true; + const compactionPayload = compactionTurn ? runnerdCompactionTurnPayload(this.options.provider ?? "codex") : null; const input = Array.isArray(params.input) ? params.input.map(record) : []; const message = input .filter((item) => item.type !== "skill") @@ -6673,7 +6690,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { const startResult = await this.#commandResult( "turn.start", { - text: message, + ...(compactionPayload ?? { text: message }), ...(this.options.recordTimingReceipt ? {performance:true} : {}), ...((this.options.provider ?? "codex") === "codex" && params.effort !== undefined ? { effort: params.effort } : {}), ...(reasoningMode === undefined ? {} : { reasoningMode }), diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index df66c1fade0..2f9fc42456c 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -1,3 +1,4 @@ +import { validatePrpStructuredRunResult } from "./protocol/replay-contract.js"; import { createHash } from "node:crypto"; import { writeFile } from "node:fs/promises"; import { join } from "node:path"; @@ -503,6 +504,47 @@ describe("executeNativeSession recovery", () => { expect(checkpoints.some(snapshot => snapshot.terminalTurns?.some(turn => turn.turnId === "turn-recovery"))).toBe(true); }); + it("never continues task work when a restart loses a compaction turn", async () => { + const failure = { code: "provider_turn_lost_on_restore", recoverable: true }; + const checkpoint: PersistedNativeSession = { + backendKind: "runner", driverKind: "codex_app_server", sessionId: "thread-recovery", identity, + providerSessionId: "provider-recovery", cursor: "0", activeTurnId: "turn-recovery", + terminalTurns: [], semanticResult: null, pendingRuntimeRequests: [], lineage: [], + }; + const interrupted: PersistedNativeSession = { + ...checkpoint, activeTurnId: null, + terminalTurns: [{ turnId: "turn-recovery", fingerprint: JSON.stringify({ terminalState: "failed", error: failure, result: null }) }], + }; + const startTurn = vi.fn(async () => ({ turnId: "turn-continuation" })); + const session: NativeSession = { + identity: () => identity, + async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + async *events() { yield runnerEvent(1, "turn.failed", { status: "failed", error: failure }); }, + startTurn, + async result() { return null; }, + async snapshot() { return structuredClone(interrupted); }, + async close() {}, + }; + const backend: NativeSessionBackend = { + async descriptor() { return { kind: "runner", name: "restart", version: "1", capabilities: await session.capabilities() }; }, + async openSession() { throw new Error("must preserve the provider session"); }, + async recoverSession() { return { recovered: true, session }; }, + }; + const port: ControlPlanePort = { + async openRun() {}, + async checkpointSession() {}, + async appendEvent(event) { return { cursor: event.sourceSeq, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; }, + async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; }, + async completeRun() {}, + }; + await expect(executeNativeSession({ + input, backend, controlPlane: port, persistedSession: checkpoint, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", + resumeInterruptedTurn: true, sessionCompaction: { requestId: "compact-1" }, + })).rejects.toMatchObject({ code: "native_provider_terminal_failed" }); + expect(startTurn).not.toHaveBeenCalled(); + }); + it.each([undefined, 0, 7 * 24 * 60 * 60 * 1000, 30 * 24 * 60 * 60 * 1000])( "honors long-lived turn duration independently of operation bounds (%s)", async (turnTimeoutMs) => { @@ -1188,6 +1230,189 @@ describe("executeNativeSession recovery", () => { }); }); + describe("context compaction", () => { + const capabilities = { + resume: true, + typedEvents: true, + steering: false, + interruption: true, + structuredResult: true, + }; + const port: ControlPlanePort = { + async openRun() {}, + async checkpointSession() {}, + async appendEvent(event) { + return { cursor: event.sourceSeq, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; + }, + async replayEvents() { + return { events: [], highestContiguousSourceSeq: 0 }; + }, + async completeRun() {}, + }; + function compactingSession(startTurn: NativeSession["startTurn"], activeTurnId: string | null = null): NativeSession { + return { + identity: () => identity, + async capabilities() { + return capabilities; + }, + async *events() { + yield runnerEvent(1, "turn.started"); + yield runnerEvent(2, "context.compacted", { + schema: "paperclip.context.compacted.v1", + compactionId: "compaction-1", + reason: "provider", + preTokens: null, + postTokens: null, + sameSession: true, + }); + yield runnerEvent(3, "turn.completed"); + }, + startTurn, + async result() { + return null; + }, + async snapshot() { + return { + backendKind: "mock", + sessionId: "driver-recovery", + identity, + providerSessionId: "provider-recovery", + cursor: null, + activeTurnId, + pendingRuntimeRequests: [], + lineage: [], + }; + }, + close: vi.fn(async () => {}), + }; + } + const persistedSession: PersistedNativeSession = { + backendKind: "mock", + sessionId: "driver-recovery", + identity, + providerSessionId: "provider-recovery", + cursor: null, + activeTurnId: null, + pendingRuntimeRequests: [], + lineage: [], + }; + + it("compacts a resumed session and settles without task work", async () => { + const startTurn = vi.fn(async () => ({ turnId: "turn-recovery" })); + const session = compactingSession(startTurn); + const backend: NativeSessionBackend = { + async descriptor() { + return { kind: "mock", name: "compaction", version: "1", capabilities }; + }, + async openSession() { + throw new Error("fresh session must not be opened"); + }, + async recoverSession() { + return { recovered: true, session }; + }, + }; + const completed = await executeNativeSession({ + input, + backend, + controlPlane: port, + runnerInstanceId: "runner-recovery", + controlPlaneInstanceId: "control-recovery", + persistedSession, + requireSessionCloseBeforeReturn: true, + sessionCompaction: { requestId: "compact-1" }, + }); + expect(startTurn).toHaveBeenCalledTimes(1); + expect(startTurn).toHaveBeenCalledWith(expect.objectContaining({ compact: true, continuation: true })); + expect(completed.result).toMatchObject({ + reportedWorkDisposition: "yielded", + summary: "Context compacted on request.", + completionClaim: { objectiveSatisfied: false, remainingWork: [] }, + }); + expect(completed.result.continuation).toMatchObject({ kind: "same_agent" }); + expect(validatePrpStructuredRunResult(completed.result)).toMatchObject({ ok: true }); + }); + + it("keeps waiting on its own compaction turn after a restart", async () => { + const startTurn = vi.fn(async () => ({ turnId: "turn-recovery" })); + const session = compactingSession(startTurn, "turn-recovery"); + const backend: NativeSessionBackend = { + async descriptor() { + return { kind: "mock", name: "compaction", version: "1", capabilities }; + }, + async openSession() { + throw new Error("fresh session must not be opened"); + }, + async recoverSession() { + return { recovered: true, session }; + }, + }; + const completed = await executeNativeSession({ + input, + backend, + controlPlane: port, + runnerInstanceId: "runner-recovery", + controlPlaneInstanceId: "control-recovery", + persistedSession: { ...persistedSession, activeTurnId: "turn-recovery" }, + requireSessionCloseBeforeReturn: true, + sessionCompaction: { requestId: "compact-1" }, + }); + expect(startTurn).not.toHaveBeenCalled(); + expect(completed.result).toMatchObject({ summary: "Context compacted on request." }); + }); + + it("never compacts over another run's checkpoint", async () => { + const startTurn = vi.fn(async () => ({ turnId: "turn-recovery" })); + const session = compactingSession(startTurn, "turn-recovery"); + const backend: NativeSessionBackend = { + async descriptor() { + return { kind: "mock", name: "compaction", version: "1", capabilities }; + }, + async openSession() { + throw new Error("fresh session must not be opened"); + }, + async recoverSession() { + return { recovered: true, session }; + }, + }; + await expect(executeNativeSession({ + input, + backend, + controlPlane: port, + runnerInstanceId: "runner-recovery", + controlPlaneInstanceId: "control-recovery", + persistedSession: { + ...persistedSession, + identity: { ...identity, runId: "run-earlier" }, + activeTurnId: "turn-recovery", + }, + sessionCompaction: { requestId: "compact-1" }, + })).rejects.toThrow("native_session_checkpoint_binding_mismatch"); + expect(startTurn).not.toHaveBeenCalled(); + }); + + it("refuses to compact a fresh session that has no prior context", async () => { + const startTurn = vi.fn(async () => ({ turnId: "turn-recovery" })); + const session = compactingSession(startTurn); + const backend: NativeSessionBackend = { + async descriptor() { + return { kind: "mock", name: "compaction", version: "1", capabilities }; + }, + async openSession() { + return session; + }, + }; + await expect(executeNativeSession({ + input, + backend, + controlPlane: port, + runnerInstanceId: "runner-recovery", + controlPlaneInstanceId: "control-recovery", + sessionCompaction: { requestId: "compact-1" }, + })).rejects.toThrow("native_context_compaction_unavailable"); + expect(startTurn).not.toHaveBeenCalled(); + }); + }); + it.each([ { recoverable: false, diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 13f222af23e..273f84ba33a 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -230,6 +230,12 @@ export interface ExecuteNativeSessionOptions { sessionGoalControl?: NativeSessionGoalControl | null; /** Resume the active durable session goal after a bounded heartbeat rollover. */ resumeSessionGoalHeartbeat?: boolean; + /** + * Compact the resumed provider session's context instead of starting an + * ordinary turn. The run settles with a synthesized result once the + * provider's compaction turn completes; it does no task work. + */ + sessionCompaction?: { requestId: string; contextTokensBefore?: number } | null; /** * Wait for the backend's close contract before returning a durable result. * Use this only for backends whose close path is internally bounded and @@ -901,6 +907,8 @@ async function consumeTurn( resumeInterruptedTurn?: (event: PrpEvent, signal: AbortSignal) => Promise, checkpointGovernedWait?: (wait: NonNullable, signal: AbortSignal) => Promise, initialGovernedWait?: PersistedNativeSession["governedWait"], + /** Settles a compaction-only turn, which reports no semantic result. */ + compactionResult?: PrpStructuredRunResult, ) { let timer: ReturnType | undefined; const appendAbort = new AbortController(); @@ -1289,6 +1297,14 @@ async function consumeTurn( } if (isTurnTerminal(event)) { if (providerFailure) throw providerFailure; + if (compactionResult && event.eventType === "turn.completed") { + return { + event, + eventCount, + highestContiguousSourceSeq, + governedResult: compactionResult, + }; + } if (semanticResultProposal === null && governedResult === null && inputTimers.size === 0 && await resumeInterruptedTurn?.(event, appendAbort.signal)) continue; return { @@ -1534,6 +1550,43 @@ export async function applyNativeSessionGoalControl( }); } +/** + * Result for a compaction-only run. Compaction does no task work, so the + * result yields back to the same agent; the server's status arbiter preserves + * the task status and schedules nothing for a compaction run. + */ +export function sessionCompactionResult( + input: NativeExecutionInput, + requestId: string, +): PrpStructuredRunResult { + const summary = "Context compacted on request."; + return { + schema: "paperclip.run_result.v1", + reportedWorkDisposition: "yielded", + summary, + completionClaim: { + contractRevision: input.completionContract.contract.revision, + objectiveSatisfied: false, + criteria: input.completionContract.contract.criteria.map((criterion) => ({ + criterionId: criterion.id, + status: "unknown", + evidenceRefs: [], + explanation: "A context compaction run does no task work.", + })), + remainingWork: [], + }, + evidence: [], + verification: [], + attentionRequests: [], + artifacts: [], + continuation: { + kind: "same_agent", + summary: "The agent's context was compacted; the task continues unchanged.", + idempotencyKey: `context-compaction:${input.binding.issueId}:${requestId}`, + }, + }; +} + function sessionGoalResult( input: NativeExecutionInput, goal: HarnessThreadGoal | null, @@ -2394,7 +2447,8 @@ export async function executeNativeSession( let restartContinuationStarted = false; const resumeInterruptedTurn = async (event: PrpEvent, signal: AbortSignal) => { if (restartContinuationStarted || !recovered || !options.resumeInterruptedTurn || - input.provider.kind !== "codex" || options.sessionGoalControl || options.resumeSessionGoalHeartbeat || + // A compaction-only run never continues task work after a restart. + input.provider.kind !== "codex" || options.sessionCompaction || options.sessionGoalControl || options.resumeSessionGoalHeartbeat || recoveredSnapshot.goal || persistedSession?.semanticResult || persistedSession?.pendingRuntimeRequests?.length || event.eventType !== "turn.failed" || !isNativeRestartInterruption(event.payload.error) || event.turnId !== persistedSession?.activeTurnId) return false; @@ -2571,13 +2625,19 @@ export async function executeNativeSession( await checkpoint(signal); }, governedWait, + options.sessionCompaction + ? sessionCompactionResult(input, options.sessionCompaction.requestId) + : undefined, ) : Promise.resolve({ event: recoveryTerminal, eventCount: 0, highestContiguousSourceSeq: replayedDisposition === null ? 0 : recoveryTerminal.sourceSeq, - governedResult: null, + // A restarted compaction run adopts its completed compaction turn. + governedResult: options.sessionCompaction && recoveryTerminal.eventType === "turn.completed" + ? sessionCompactionResult(input, options.sessionCompaction.requestId) + : null, }); // Event consumption must begin before startTurn so an eager provider cannot // outrun us. Observe its rejection immediately, though: if startTurn or @@ -2592,7 +2652,28 @@ export async function executeNativeSession( !adoptedProviderTerminal && !checkpointedDispositionTerminal && !dispositionRecoveryStillOwned)); - if (options.sessionGoalControl) { + if (options.sessionCompaction) { + // Compaction summarizes existing provider context. A new session or + // one that is still running a turn has nothing safe to compact. + // A restart of this same compaction run keeps waiting on (or adopts) + // the compaction turn it already started instead of starting another. + const resumingThisCompaction = recovered && !shouldStartFreshTurn && + persistedSession?.identity.runId === input.binding.runId; + if (!recovered || (!shouldStartFreshTurn && !resumingThisCompaction)) { + throw new Error( + "native_context_compaction_unavailable: the agent has no idle resumed session to compact", + ); + } + if (shouldStartFreshTurn) await session.startTurn({ + message: { role: "user", text: "" }, + continuation: true, + compact: true, + ...(options.sessionCompaction.contextTokensBefore === undefined + ? {} + : { contextTokensBefore: options.sessionCompaction.contextTokensBefore }), + }); + await checkpoint(); + } else if (options.sessionGoalControl) { // Explicit controls remain authoritative after controller loss. In // particular, pause/clear/edit must reach a recovered session even // while its provider turn is still active. diff --git a/packages/paperclip-runner/src/provider-events.ts b/packages/paperclip-runner/src/provider-events.ts index 850fb4b125b..cb321859119 100644 --- a/packages/paperclip-runner/src/provider-events.ts +++ b/packages/paperclip-runner/src/provider-events.ts @@ -1193,10 +1193,11 @@ export function canonicalProviderEventsFromCodex( }, ]; } + // A compaction item starts before the summary exists; only its completion + // (or the legacy thread/compacted notification) marks the compaction. if ( method === "thread/compacted" || - ((method === "item/completed" || method === "item/started") && - type === "contextCompaction") + (method === "item/completed" && type === "contextCompaction") ) { return [ { diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index b73dfa295ff..4f351647499 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -22,7 +22,7 @@ export { type RunnerGoalActionRequest, type RunnerGoalActionAccepted, } from "./runner-goal.js"; -export type { IssueContextUsage } from "./issue-context-usage.js"; +export type { IssueContextCompaction, IssueContextUsage } from "./issue-context-usage.js"; export { ADAPTER_AUTH_MISSING_CHECK_CODE } from "./adapter-auth-check-code.js"; export { CONNECTION_INTENT_AGENT_GUIDANCE, diff --git a/packages/shared/src/issue-context-usage.ts b/packages/shared/src/issue-context-usage.ts index 9e4641deb2b..2de0de824da 100644 --- a/packages/shared/src/issue-context-usage.ts +++ b/packages/shared/src/issue-context-usage.ts @@ -20,4 +20,16 @@ export interface IssueContextUsage { costUsd: number | null; /** Compactions recorded for this session. */ compactions: number; + /** Whether the board can ask the agent to compact now. */ + compaction?: IssueContextCompaction; +} + +/** "Compact now" state for the task's assignee session. */ +export interface IssueContextCompaction { + status: "available" | "pending" | "unavailable"; + reasonCode: "no_assignee" | "task_closed" | "unsupported" | "agent_busy" | null; + /** Why compaction is unavailable, in words the board can act on. */ + reason: string | null; + /** The queued or running compaction run while pending. */ + runId?: string; } diff --git a/server/src/__tests__/context-compaction.test.ts b/server/src/__tests__/context-compaction.test.ts new file mode 100644 index 00000000000..3d618ebcd23 --- /dev/null +++ b/server/src/__tests__/context-compaction.test.ts @@ -0,0 +1,106 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { agents, companies, createDb, heartbeatRuns, issues } from "@paperclipai/db"; +import { + ContextCompactionError, + contextCompactionService, + contextCompactionSupport, + isContextCompactionRun, + readContextCompactionRequest, +} from "../services/context-compaction.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +describe("context compaction support", () => { + it("supports Codex and Claude on the Paperclip runner only", () => { + expect(contextCompactionSupport({ adapterType: "paperclip_runner", adapterConfig: { provider: "codex" } }).supported).toBe(true); + expect(contextCompactionSupport({ adapterType: "paperclip_runner", adapterConfig: {} }).supported).toBe(true); + expect(contextCompactionSupport({ adapterType: "paperclip_runner", adapterConfig: { provider: "acpx", acpxAgent: "claude" } }).supported).toBe(true); + expect(contextCompactionSupport({ adapterType: "paperclip_runner", adapterConfig: { provider: "acpx", acpxAgent: "pi" } }).supported).toBe(false); + expect(contextCompactionSupport({ adapterType: "paperclip_runner", adapterConfig: { provider: "opencode" } }).supported).toBe(false); + expect(contextCompactionSupport({ adapterType: "codex_local", adapterConfig: {} }).supported).toBe(false); + }); + + it("recognizes a compaction run from its context", () => { + expect(readContextCompactionRequest({ contextCompactionRequestId: "req-1" })).toEqual({ requestId: "req-1" }); + expect(isContextCompactionRun({ contextCompactionRequestId: " " })).toBe(false); + expect(isContextCompactionRun({ goalControlRequestId: "goal-1" })).toBe(false); + expect(isContextCompactionRun(null)).toBe(false); + }); +}); + +const support = await getEmbeddedPostgresTestSupport(); +const suite = support.supported ? describe : describe.skip; + +suite("context compaction availability", () => { + let db: ReturnType; + let temporaryDb: Awaited>; + beforeAll(async () => { + temporaryDb = await startEmbeddedPostgresTestDatabase("context-compaction-"); + db = createDb(temporaryDb.connectionString); + }, 60_000); + afterAll(async () => { await temporaryDb?.cleanup(); }); + + async function fixture(adapterConfig: Record = { provider: "codex" }, adapterType = "paperclip_runner") { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Compaction fixture", issuePrefix: `C${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Codie", role: "engineer", status: "idle", adapterType, adapterConfig }); + await db.insert(issues).values({ id: issueId, companyId, title: "Compact", status: "in_progress", assigneeAgentId: agentId }); + return { companyId, agentId, issue: { id: issueId, assigneeAgentId: agentId as string | null, status: "in_progress" } }; + } + + it("is available for an idle supported agent and queues a compaction run", async () => { + const { companyId, agentId, issue } = await fixture(); + const wakeup = vi.fn(async () => ({ id: randomUUID() })); + const service = contextCompactionService(db, { wakeup }); + expect(await service.availability(companyId, issue)).toEqual({ status: "available", reasonCode: null, reason: null }); + const accepted = await service.request({ companyId, issue, userId: "board-user" }); + expect(wakeup).toHaveBeenCalledWith(agentId, expect.objectContaining({ + reason: "context_compaction", + requestedByActorType: "user", + requestedByActorId: "board-user", + allowRunCoalescing: false, + manualUserWake: true, + contextSnapshot: expect.objectContaining({ + issueId: issue.id, + contextCompactionRequestId: accepted.requestId, + skipIssueComment: true, + }), + })); + }); + + it("waits for an active run and reports a compaction already underway", async () => { + const { companyId, agentId, issue } = await fixture(); + const service = contextCompactionService(db, { wakeup: vi.fn() }); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, scopeKind: "issue", issueId: issue.id, status: "running" }); + expect(await service.availability(companyId, issue)).toMatchObject({ status: "unavailable", reasonCode: "agent_busy" }); + await expect(service.request({ companyId, issue, userId: "board-user" })).rejects.toMatchObject({ code: "agent_busy" }); + + const { companyId: otherCompanyId, agentId: otherAgentId, issue: otherIssue } = await fixture(); + const compactionRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: compactionRunId, companyId: otherCompanyId, agentId: otherAgentId, scopeKind: "issue", status: "queued", + contextSnapshot: { issueId: otherIssue.id, contextCompactionRequestId: "req-1" }, + }); + expect(await service.availability(otherCompanyId, otherIssue)).toEqual({ + status: "pending", reasonCode: null, reason: null, runId: compactionRunId, + }); + await expect(service.request({ companyId: otherCompanyId, issue: otherIssue, userId: "board-user" })) + .rejects.toBeInstanceOf(ContextCompactionError); + }); + + it("explains unsupported harnesses, closed tasks, and missing assignees", async () => { + const unsupported = await fixture({ provider: "opencode" }); + const service = contextCompactionService(db); + expect(await service.availability(unsupported.companyId, unsupported.issue)).toMatchObject({ + status: "unavailable", reasonCode: "unsupported", + }); + const closed = await fixture(); + expect(await service.availability(closed.companyId, { ...closed.issue, status: "done" })).toMatchObject({ + status: "unavailable", reasonCode: "task_closed", + }); + expect(await service.availability(closed.companyId, { ...closed.issue, assigneeAgentId: null })).toMatchObject({ + status: "unavailable", reasonCode: "no_assignee", + }); + }); +}); diff --git a/server/src/__tests__/heartbeat-auto-checkout.test.ts b/server/src/__tests__/heartbeat-auto-checkout.test.ts index 035d1456ae4..a422ae5780d 100644 --- a/server/src/__tests__/heartbeat-auto-checkout.test.ts +++ b/server/src/__tests__/heartbeat-auto-checkout.test.ts @@ -10,6 +10,7 @@ describe("shouldAutoCheckoutIssueForWake", () => { { contextSnapshot: {} }, { contextSnapshot: { wakeReason: "issue_comment_mentioned" } }, { contextSnapshot: { wakeReason: "source_scoped_recovery_action" } }, { contextSnapshot: { wakeReason: "execution_resumed" } }, + { contextSnapshot: { wakeReason: "context_compaction", contextCompactionRequestId: "req-1" } }, ])("does not grant an automatic checkout outside the eligible assigned wake: %j", override => { expect(shouldAutoCheckoutIssueForWake({ contextSnapshot: { wakeReason: "issue_assigned" }, issueStatus: "todo", diff --git a/server/src/__tests__/issue-context-usage.test.ts b/server/src/__tests__/issue-context-usage.test.ts index 91eef26a181..b7616e581ce 100644 --- a/server/src/__tests__/issue-context-usage.test.ts +++ b/server/src/__tests__/issue-context-usage.test.ts @@ -44,6 +44,9 @@ suite("issue context usage", () => { createdAt: new Date("2026-11-05T10:00:00Z") }, { ...base, runId, seq: 2, eventType: "context.compacted", payload: prp(runId, "session-a", "context.compacted", { schema: "paperclip.context.compacted.v1", compactionId: "c-1" }), createdAt: new Date("2026-11-05T10:00:01Z") }, + // The same compaction reported twice (item completion and thread/compacted) counts once. + { ...base, runId, seq: 5, eventType: "context.compacted", payload: prp(runId, "session-a", "context.compacted", { + schema: "paperclip.context.compacted.v1", compactionId: "c-1" }), createdAt: new Date("2026-11-05T10:00:01Z") }, { ...base, runId, seq: 3, eventType: "item.completed", payload: prp(runId, "session-a", "item.completed", usage(38_200)), createdAt: new Date("2026-11-05T10:00:02Z") }, { ...base, runId, seq: 4, eventType: "item.completed", payload: prp(runId, "session-a", "item.completed", { kind: "usage", usage: { total: {} } }), diff --git a/server/src/modules/wake-queue/domain/context.test.ts b/server/src/modules/wake-queue/domain/context.test.ts index 9dc44b8c099..08b6b54738b 100644 --- a/server/src/modules/wake-queue/domain/context.test.ts +++ b/server/src/modules/wake-queue/domain/context.test.ts @@ -144,3 +144,14 @@ describe("enrichPromotedWakeContext", () => { expect(queued.contextSnapshot).toMatchObject({ issueId: "issue-1", taskId: "task-1", taskKey: "TASK-1" }); }); }); + +describe("context compaction wakes", () => { + it("never merge with ordinary wakes, in either order", async () => { + const { wakeMustNotJoinActiveRun } = await import("./context.js"); + const compaction = { issueId: "issue-1", contextCompactionRequestId: "req-1" }; + const comment = { issueId: "issue-1", wakeCommentId: "comment-1" }; + expect(wakeMustNotJoinActiveRun(compaction, comment, false)).toBe(true); + expect(wakeMustNotJoinActiveRun(comment, compaction, false)).toBe(true); + expect(wakeMustNotJoinActiveRun(comment, { issueId: "issue-1", wakeCommentId: "comment-2" }, false)).toBe(false); + }); +}); diff --git a/server/src/modules/wake-queue/domain/context.ts b/server/src/modules/wake-queue/domain/context.ts index a4f0df74e52..ab2d4d5d060 100644 --- a/server/src/modules/wake-queue/domain/context.ts +++ b/server/src/modules/wake-queue/domain/context.ts @@ -81,7 +81,11 @@ export function wakeMustNotJoinActiveRun( wakeContext: Record, interactionReceipt: boolean, ): boolean { + // A context compaction run does no task work, so ordinary wakes and + // compactions never merge, in either arrival order. return interactionReceipt || Boolean(parseObject(activeRunContext).interactionId) || + Boolean(readNonEmptyString(parseObject(activeRunContext).contextCompactionRequestId)) || + Boolean(readNonEmptyString(wakeContext.contextCompactionRequestId)) || wakeChatConversationsConflict(activeRunContext, wakeContext); } diff --git a/server/src/routes/activity.ts b/server/src/routes/activity.ts index 8fd095361d3..d1642e02d43 100644 --- a/server/src/routes/activity.ts +++ b/server/src/routes/activity.ts @@ -11,6 +11,7 @@ import { badRequest, forbidden } from "../errors.js"; import { agentActionAuditService } from "../services/agent-action-audit.js"; import { logActivity } from "../services/activity-log.js"; import { issueContextUsage } from "../services/issue-context-usage.js"; +import { contextCompactionService } from "../services/context-compaction.js"; import { canActorReadHeartbeatRun } from "../services/heartbeat-run-privacy.js"; import { issueReadSqlCondition } from "../services/authorization.js"; import { caseActivityReadSqlCondition } from "../services/case-activity-access.js"; @@ -392,7 +393,10 @@ export function activityRoutes(db: Db) { const issue = await getAccessibleResource(req, res, resolveIssueByRef(req.params.id as string), "Issue not found"); if (!issue) return; if (!(await assertIssueReadAllowed(req, res, issue))) return; - res.json(await issueContextUsage(db, issue.companyId, issue)); + const usage = await issueContextUsage(db, issue.companyId, issue); + res.json(usage + ? { ...usage, compaction: await contextCompactionService(db).availability(issue.companyId, issue) } + : null); }); router.get("/heartbeat-runs/:runId/issues", async (req, res) => { diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 16e12df9d1e..7d5cc8835f6 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -229,6 +229,8 @@ import { RunnerGoalActionError, RunnerGoalConflictError, } from "../services/runner-goals.js"; +import { ContextCompactionError, contextCompactionService } from "../services/context-compaction.js"; +import { issueContextUsage } from "../services/issue-context-usage.js"; import { queueLiveRunnerPrpCommand } from "../realtime/runner-prp-ws.js"; import { questionResponseDeliveryService } from "../services/question-response-delivery.js"; import { emitAgentTaskRunById } from "../services/agent-task-run-telemetry.js"; @@ -3534,6 +3536,9 @@ export function issueRoutes( resolveNativeQuestion: (interaction) => deliverNativeQuestionResponse(db, interaction), }); + const contextCompaction = contextCompactionService(db, { + wakeup: (agentId, options) => heartbeat.wakeup(agentId, options), + }); const runnerGoals = runnerGoalService(db, { enqueueOfflineControl: async ({ issueId, agentId, requestId, control }) => { const run = await heartbeat.wakeup(agentId, { @@ -9290,6 +9295,60 @@ export function issueRoutes( }, ); + router.post("/issues/:id/context/compact", async (req, res) => { + assertBoard(req); + if (!req.actor.userId) throw forbidden("Board user context required"); + const issue = await getAccessibleResource( + req, + res, + getIssueById(req, req.params.id as string), + "Issue not found", + ); + if (!issue) return; + const decision = await decideIssueAccess(req, issue, "issue:mutate"); + if (!decision.allowed) throw forbidden(decision.explanation, authorizationDeniedDetails(decision)); + // An Agent Chat run starts only for the conversation's owner. + if (issue.conversationAgentId) { + if (!(await instanceSettings.getExperimental()).enableAgentChat) throw notFound("Agent Chat is disabled"); + if (req.actor.userId !== issue.conversationUserId) + throw forbidden("Only the conversation owner can compact a chat agent's context"); + } + try { + const reading = await issueContextUsage(db, issue.companyId, issue); + const accepted = await contextCompaction.request({ + companyId: issue.companyId, + issue, + userId: req.actor.userId, + contextTokensBefore: reading?.usedTokens ?? null, + }); + const actor = getActorInfo(req); + await logActivity(db, { + companyId: issue.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + runId: actor.runId, + action: "issue.context_compaction_requested", + entityType: "issue", + entityId: issue.id, + issueId: issue.id, + details: { + requestId: accepted.requestId, + compactionRunId: accepted.runId, + assigneeAgentId: issue.assigneeAgentId, + }, + }); + res.status(202).json(accepted); + } catch (error) { + if (error instanceof ContextCompactionError) { + const status = error.code === "agent_busy" || error.code === "already_pending" ? 409 : 422; + res.status(status).json({ error: error.message, code: error.code }); + return; + } + throw error; + } + }); + router.get("/issues/:id/heartbeat-context", async (req, res) => { const id = req.params.id as string; const issue = await getAccessibleResource( diff --git a/server/src/routes/openapi.ts b/server/src/routes/openapi.ts index 3669aaf381b..09f23d3eb92 100644 --- a/server/src/routes/openapi.ts +++ b/server/src/routes/openapi.ts @@ -6178,6 +6178,16 @@ registry.registerPath({ responses: { 200: r.ok(), 401: r.unauthorized, 403: r.forbidden, 404: r.notFound }, }); +registry.registerPath({ + method: "post", + path: "/api/issues/{id}/context/compact", + tags: ["issues"], + summary: "Ask the assignee agent to compact its context window", + description: "Board only. Queues a short run that resumes the assignee's runner session and compacts it (Codex and Claude on the Paperclip runner). Unavailable while the agent is running on the task.", + request: { params: z.object({ id: z.string() }) }, + responses: { 202: r.ok(), 401: r.unauthorized, 403: r.forbidden, 404: r.notFound, 409: r.conflict, 422: r.unprocessable }, +}); + registry.registerPath({ method: "get", path: "/api/heartbeat-runs/{runId}/issues", diff --git a/server/src/services/context-compaction.ts b/server/src/services/context-compaction.ts new file mode 100644 index 00000000000..00eccfae0f2 --- /dev/null +++ b/server/src/services/context-compaction.ts @@ -0,0 +1,201 @@ +import { randomUUID } from "node:crypto"; +import { and, desc, eq, inArray, or, sql } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { agents, heartbeatRuns } from "@paperclipai/db"; +import type { IssueContextCompaction } from "@paperclipai/shared"; + +/** + * On-demand context compaction ("Compact now"). + * + * Agents run with a per-turn session lifecycle: the harness session exists + * only during a run. A compaction therefore runs as its own short heartbeat + * run that resumes the assignee's session, asks the harness to compact (Codex + * `thread/compact/start`, Claude's ACP `/compact` command), and settles + * without changing the task. The run is marked by + * `contextSnapshot.contextCompactionRequestId`. + */ + +export const CONTEXT_COMPACTION_WAKE_REASON = "context_compaction"; + +const ACTIVE_RUN_STATUSES = ["queued", "scheduled_retry", "running"] as const; + +type AgentBinding = Pick; + +function asRecord(value: unknown): Record { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? value as Record + : {}; +} + +function nonEmptyString(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value : null; +} + +/** The compaction request a run was admitted for, if it is a compaction run. */ +export function readContextCompactionRequest( + context: unknown, +): { requestId: string; contextTokensBefore?: number } | null { + const record = asRecord(context); + const requestId = nonEmptyString(record.contextCompactionRequestId); + if (!requestId) return null; + const before = record.contextCompactionTokensBefore; + return typeof before === "number" && Number.isSafeInteger(before) && before >= 0 + ? { requestId, contextTokensBefore: before } + : { requestId }; +} + +export function isContextCompactionRun(context: unknown): boolean { + return readContextCompactionRequest(context) !== null; +} + +/** + * Harnesses that can compact on request. Compaction needs the Paperclip + * runner, which resumes the provider session: Codex through its app-server, + * Claude through ACP. + */ +export function contextCompactionSupport(agent: AgentBinding): { supported: true } | { supported: false; reason: string } { + const unsupported = { supported: false as const, reason: "This agent's harness can't compact on request." }; + if (agent.adapterType !== "paperclip_runner") return unsupported; + const config = asRecord(agent.adapterConfig); + const provider = typeof config.provider === "string" ? config.provider : "codex"; + if (provider === "codex") return { supported: true }; + if (provider === "acpx" && config.acpxAgent === "claude") return { supported: true }; + return unsupported; +} + +export class ContextCompactionError extends Error { + constructor( + readonly code: "no_assignee" | "task_closed" | "unsupported" | "agent_busy" | "already_pending" | "wake_skipped", + message: string, + ) { + super(message); + } +} + +type WakeupFn = ( + agentId: string, + options: { + source: "on_demand"; + triggerDetail: "manual"; + reason: string; + payload: Record; + idempotencyKey: string; + requestedByActorType: "user"; + requestedByActorId: string; + contextSnapshot: Record; + allowRunCoalescing: false; + manualUserWake: true; + }, +) => Promise<{ id: string } | null>; + +export function contextCompactionService(db: Db, deps: { wakeup?: WakeupFn } = {}) { + /** The assignee's newest unfinished run for this task, if any. */ + async function activeRun(companyId: string, issueId: string, agentId: string) { + return db + .select({ id: heartbeatRuns.id, contextSnapshot: heartbeatRuns.contextSnapshot }) + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.agentId, agentId), + inArray(heartbeatRuns.status, [...ACTIVE_RUN_STATUSES]), + or( + eq(heartbeatRuns.issueId, issueId), + eq(heartbeatRuns.nativeIssueId, issueId), + sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, + ), + )) + .orderBy(desc(heartbeatRuns.createdAt)) + .limit(1) + .then((rows) => rows[0] ?? null); + } + + async function availability( + companyId: string, + issue: { id: string; assigneeAgentId: string | null; status: string }, + ): Promise { + if (issue.status === "done" || issue.status === "cancelled") { + return { status: "unavailable", reasonCode: "task_closed", reason: "Available while the task is open." }; + } + if (!issue.assigneeAgentId) { + return { status: "unavailable", reasonCode: "no_assignee", reason: "Assign an agent to compact its context." }; + } + const agent = await db + .select({ adapterType: agents.adapterType, adapterConfig: agents.adapterConfig }) + .from(agents) + .where(and(eq(agents.id, issue.assigneeAgentId), eq(agents.companyId, companyId))) + .then((rows) => rows[0] ?? null); + if (!agent) { + return { status: "unavailable", reasonCode: "no_assignee", reason: "Assign an agent to compact its context." }; + } + const support = contextCompactionSupport(agent); + if (!support.supported) return { status: "unavailable", reasonCode: "unsupported", reason: support.reason }; + const run = await activeRun(companyId, issue.id, issue.assigneeAgentId); + if (run && isContextCompactionRun(run.contextSnapshot)) { + return { status: "pending", reasonCode: null, reason: null, runId: run.id }; + } + if (run) { + // Compacting interrupts a turn, so wait for the current run to end. + return { + status: "unavailable", + reasonCode: "agent_busy", + reason: "Available when the agent finishes its current run.", + }; + } + return { status: "available", reasonCode: null, reason: null }; + } + + /** Queues a compaction run for the task's assignee. */ + async function request(input: { + companyId: string; + issue: { id: string; assigneeAgentId: string | null; status: string }; + userId: string; + /** The session's last context reading, shown as the divider's "before" size. */ + contextTokensBefore?: number | null; + }): Promise<{ requestId: string; runId: string }> { + const current = await availability(input.companyId, input.issue); + if (current.status === "pending") { + throw new ContextCompactionError("already_pending", "The agent is already compacting its context."); + } + if (current.status === "unavailable") { + throw new ContextCompactionError( + current.reasonCode ?? "unsupported", + current.reason ?? "Context compaction is unavailable.", + ); + } + if (!deps.wakeup) throw new Error("context compaction wakeup is not configured"); + const agentId = input.issue.assigneeAgentId!; + const requestId = randomUUID(); + const run = await deps.wakeup(agentId, { + source: "on_demand", + triggerDetail: "manual", + reason: CONTEXT_COMPACTION_WAKE_REASON, + payload: { issueId: input.issue.id, requestId, intent: CONTEXT_COMPACTION_WAKE_REASON }, + idempotencyKey: `${CONTEXT_COMPACTION_WAKE_REASON}:${requestId}`, + requestedByActorType: "user", + requestedByActorId: input.userId, + // A compaction never joins, or absorbs, another run's work. + allowRunCoalescing: false, + // An explicit board click; admission and dispatch recheck its owner. + manualUserWake: true, + contextSnapshot: { + issueId: input.issue.id, + taskKey: input.issue.id, + resumeIntent: true, + contextCompactionRequestId: requestId, + ...(typeof input.contextTokensBefore === "number" + ? { contextCompactionTokensBefore: input.contextTokensBefore } + : {}), + skipIssueComment: true, + }, + }); + if (!run) { + throw new ContextCompactionError( + "wake_skipped", + "Compaction did not start. Check that task execution is enabled for this instance and agent, then retry.", + ); + } + return { requestId, runId: run.id }; + } + + return { availability, request }; +} diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index accceff958c..6d0821d2764 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,4 +1,5 @@ import { resolveProviderApiPlacement } from "./native-runtime/provider-api-placement.js"; +import { isContextCompactionRun } from "./context-compaction.js"; import { registerDeferredReceiptWork } from "./heartbeat/deferred-receipt-work.js"; import { reconcileTaskChatRetryReceipts } from "./task-retry.js"; import { registerIssueRecoveryWork } from "./recovery/work.js"; @@ -721,6 +722,8 @@ export function shouldAutoCheckoutIssueForWake(input: { return false; } + // A compaction run does no task work, so it never claims a parked task. + if (isContextCompactionRun(input.contextSnapshot)) return false; const wakeReason = readNonEmptyString(input.contextSnapshot?.wakeReason); if (!wakeReason) return false; if (wakeReason === "issue_comment_mentioned") return false; @@ -3287,6 +3290,10 @@ export function heartbeatService( const durableGoalControlRun = readNonEmptyString(context.goalControlRequestId) !== null || context.resumeSessionGoalHeartbeat === true; + // Compaction resumes a runner session; a direct adapter would run it as a prompt. + if (isContextCompactionRun(context) && agent.adapterType !== "paperclip_runner") { + throw new Error("context_compaction_requires_paperclip_runner"); + } // Goals must use the selected durable runner, never silently convert a // direct adapter or let an old goal-control wake become a normal prompt. if (durableGoalControlRun && agent.adapterType !== "paperclip_runner") { diff --git a/server/src/services/heartbeat/conversation-events.ts b/server/src/services/heartbeat/conversation-events.ts index aed0c6d17dc..bcf755d6029 100644 --- a/server/src/services/heartbeat/conversation-events.ts +++ b/server/src/services/heartbeat/conversation-events.ts @@ -31,11 +31,27 @@ function planWriteCondition(runId: string) { ))`; } +/** + * Context readings of a run that compacted its context. The compaction divider + * reads its "before → after" sizes from the readings around the compaction. + * Runs that never compacted keep their usage out of the conversation view. + */ +function compactionUsageCondition(runId: string) { + return sql`(${heartbeatRunEvents.eventType} = 'item.completed' + and (${heartbeatRunEvents.payload} -> 'prpEvent' -> 'payload' ->> 'kind') = 'usage' + and ${heartbeatRunEvents.payload} -> 'prpEvent' -> 'payload' -> 'usage' ? 'context' + and exists ( + select 1 from heartbeat_run_events c + where c.run_id = ${runId} and c.event_type = 'context.compacted' + ))`; +} + /** * Canonical PRP conversation data, excluding ordinary historical tool activity. * Successful document writes retain a body-free boundary for saved Plan cards. * Keep request lifecycles so a cold open cannot hide a question or approval. * Keep MCP startup failures so a settled run still explains missing tools. + * Keep context compactions, with their run's context readings, for the divider. * Item identity may occur on started/completed rather than each text delta. */ export function conversationEventCondition(runId: string) { @@ -49,6 +65,10 @@ export function conversationEventCondition(runId: string) { ${heartbeatRunEvents.eventType} = 'provider.notice.recorded' and ${heartbeatRunEvents.payload} #>> '{prpEvent,payload,category}' = 'mcp_server_startup_failed' ) + or ${heartbeatRunEvents.eventType} = 'context.compacted' + or ( + ${compactionUsageCondition(runId)} + ) or ( ${heartbeatRunEvents.eventType} in ('item.started', 'item.delta', 'item.completed') and (lower(replace(coalesce(${heartbeatRunEvents.payload} #>> '{prpEvent,payload,kind}', ${heartbeatRunEvents.payload} #>> '{prpEvent,payload,item,kind}', ${heartbeatRunEvents.payload} #>> '{prpEvent,payload,item,type}', ''), '_', '')) in ('agentmessage', 'assistantmessage') diff --git a/server/src/services/heartbeat/run-completion.ts b/server/src/services/heartbeat/run-completion.ts index 76780c4e1ae..84167bae1f6 100644 --- a/server/src/services/heartbeat/run-completion.ts +++ b/server/src/services/heartbeat/run-completion.ts @@ -1,4 +1,5 @@ import { preserveAcceptedRunUsageProjectionSql } from "../usage-receipts.js"; +import { isContextCompactionRun } from "../context-compaction.js"; import { providerFailureFromError } from "../../vendor/paperclip-runner/index.js"; import { nativeTaskSessionPublicationHandled } from "../native-runtime/native-task-session-publication.js"; import { settledResponsePublicationSql } from "../native-runtime/native-response-publication-receipt.js"; @@ -987,6 +988,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC readNonEmptyString( parseObject(livenessRun.contextSnapshot).goalControlRequestId, ) !== null || + isContextCompactionRun(livenessRun.contextSnapshot) || parseObject(livenessRun.contextSnapshot) .resumeSessionGoalHeartbeat === true, }); @@ -1570,6 +1572,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC readNonEmptyString( parseObject(livenessRun.contextSnapshot).goalControlRequestId, ) !== null || + isContextCompactionRun(livenessRun.contextSnapshot) || parseObject(livenessRun.contextSnapshot) .resumeSessionGoalHeartbeat === true, }).catch((releaseError) => { diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 70ac97faf98..28fefdb965b 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1,4 +1,5 @@ import { readNativeStartupCleanup } from "../native-runtime/native-startup-cleanup.js"; +import { isContextCompactionRun } from "../context-compaction.js"; import { notifyStrandedAgentStatus } from "../live-run-recovery-notifications.js"; import { notifyRunDispatch } from "../run-dispatch-notifications.js"; import { wakeWorkspaceBusyWaitersInTx } from "../../modules/run-dispatch/index.js"; @@ -1515,7 +1516,8 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl const context = parseObject(run.contextSnapshot); if ( readNonEmptyString(context.goalControlRequestId) || - context.resumeSessionGoalHeartbeat === true + context.resumeSessionGoalHeartbeat === true || + isContextCompactionRun(context) ) return; const livenessState = run.livenessState as RunLivenessState | null; @@ -1723,6 +1725,8 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl ) { if (run.status !== "succeeded") return; const context = parseObject(run.contextSnapshot); + // A compaction run did no task work, so it hands nothing off. + if (isContextCompactionRun(context)) return; const issueId = readNonEmptyString(context.issueId) ?? readNonEmptyString(context.taskId); if (!issueId) return; @@ -2040,7 +2044,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl run: typeof heartbeatRuns.$inferSelect, ) { const contextSnapshot = parseObject(run.contextSnapshot); - if (readNonEmptyString(contextSnapshot.goalControlRequestId)) return; + if (readNonEmptyString(contextSnapshot.goalControlRequestId) || isContextCompactionRun(contextSnapshot)) return; const issueId = readNonEmptyString(contextSnapshot.issueId) ?? readNonEmptyString(contextSnapshot.taskId); @@ -2615,7 +2619,11 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl const contextSnapshot = parseObject(run.contextSnapshot); // The explicit receipt admitted one turn. A prose-only follow-up cannot // reuse it or renew it under the separate transient-failure retry policy. - if (readNonEmptyString(contextSnapshot.goalControlRequestId) || contextSnapshot.explicitUserContinuation) { + if ( + readNonEmptyString(contextSnapshot.goalControlRequestId) || + isContextCompactionRun(contextSnapshot) || + contextSnapshot.explicitUserContinuation + ) { if (run.issueCommentStatus !== "not_applicable") { await patchRunIssueCommentStatus(run.id, { issueCommentStatus: "not_applicable", diff --git a/server/src/services/heartbeat/runtime-execution.ts b/server/src/services/heartbeat/runtime-execution.ts index faee8883e49..396814abde9 100644 --- a/server/src/services/heartbeat/runtime-execution.ts +++ b/server/src/services/heartbeat/runtime-execution.ts @@ -1,4 +1,5 @@ import { readClaimedNativeStopRecovery } from "../native-runtime/native-cancellation-request.js"; +import { readContextCompactionRequest } from "../context-compaction.js"; import { remoteWorkspaceFailureCause } from "../remote-workspace-transport.js"; import { readNativePostResultFailureDiagnostic } from "../native-runtime/native-post-result-failure.js"; import { NativeStartupCleanupPendingError } from "../native-runtime/native-startup-cleanup.js"; @@ -634,6 +635,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe useRunnerd: agent.adapterType === "paperclip_runner", adapterType: agent.adapterType, sessionGoalControl, + sessionCompaction: readContextCompactionRequest(context), resumeSessionGoalHeartbeat: context.resumeSessionGoalHeartbeat === true || completedGoalControl, diff --git a/server/src/services/issue-context-usage.ts b/server/src/services/issue-context-usage.ts index 837b76b1967..e940b188e28 100644 --- a/server/src/services/issue-context-usage.ts +++ b/server/src/services/issue-context-usage.ts @@ -87,9 +87,12 @@ export async function issueContextUsage( const windowTokens = count(context?.windowTokens); if (usedTokens === null || windowTokens === null || windowTokens < 1) return null; - // Every compaction of the session, across all of its runs. + // Every compaction of the session, across all of its runs. A provider can + // report one compaction through more than one notification; count each id once. const [compactions] = await db - .select({ count: sql`count(*)::int` }) + .select({ + count: sql`count(distinct coalesce(${heartbeatRunEvents.payload} -> 'prpEvent' -> 'payload' ->> 'compactionId', ${heartbeatRunEvents.id}::text))::int`, + }) .from(heartbeatRunEvents) .where(and( eq(heartbeatRunEvents.companyId, companyId), diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 25cfc7889ee..f71a8fdd1fb 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -1,4 +1,5 @@ import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; +import { isContextCompactionRun } from "../context-compaction.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js"; import { resolveCommittedNativeRecoveryNotices } from "./native-recovery-notices.js"; @@ -1543,6 +1544,7 @@ export async function finalizeNativeRun(input: { boardResponseWaitOrigin: boardResponseWaitOrigin !== null, isConversation: isConversation(authoritativeIssue), hasActivePauseHold: pauseHold !== null, + contextCompactionRun: isContextCompactionRun(run.contextSnapshot), reviewOwnerUserId: authoritativeIssue.responsibleUserId ?? authoritativeIssue.createdByUserId ?? diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 5e9a8b38384..6013a88ba67 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -8512,6 +8512,8 @@ export async function executePaperclipNativeSession(input: { taskSessionPublication?: NativeTaskSessionPublicationContext; sessionGoalControl?: NativeSessionGoalControl | null; resumeSessionGoalHeartbeat?: boolean; + /** Compact the resumed provider session instead of running a task turn. */ + sessionCompaction?: { requestId: string; contextTokensBefore?: number } | null; preparationSpans?: NativeRunHistoricalSpan[]; /** Use a session-owned GitHub broker, rebound only after run ownership is acquired. */ managedGitHub?: boolean; @@ -10276,6 +10278,7 @@ async function executePaperclipNativeSessionWithinScope( keepSessionOpen: warmSessionId !== null, sessionGoalControl: input.sessionGoalControl, resumeSessionGoalHeartbeat: input.resumeSessionGoalHeartbeat, + sessionCompaction: input.sessionCompaction, requireSessionCloseBeforeReturn: runnerdBackend !== null || input.instructionWorkingCopy !== undefined, onSessionClosed: input.instructionWorkingCopy?.collectStopped, onCheckpoint: async (snapshot) => { diff --git a/server/src/services/native-runtime/status-arbiter.test.ts b/server/src/services/native-runtime/status-arbiter.test.ts index 109fd2c7a44..a90d98eadc3 100644 --- a/server/src/services/native-runtime/status-arbiter.test.ts +++ b/server/src/services/native-runtime/status-arbiter.test.ts @@ -77,6 +77,18 @@ describe("native status authority", () => { }); }); + it("preserves the task status after a context compaction run", () => { + for (const priorIssueStatus of ["in_progress", "in_review", "blocked", "todo"] as const) { + expect(arbitrate({ + contextCompactionRun: true, + priorIssueStatus, + assessment: assessment({ reportedDisposition: "yielded", objectiveSatisfied: false, allCriteriaSatisfied: false }), + })).toMatchObject({ + statusAction: "preserve", toStatus: priorIssueStatus, reasonCode: "context_compaction_status_preserved", effects: [], + }); + } + }); + it("keeps workspace repair ahead of an accepted case handoff", () => { expect(arbitrate({ caseHandoffAuthorized: true, workspaceFinalizeStatus: "failed" })).toMatchObject({ statusAction: "preserve", reasonCode: "finalization_failed_claim_preserved", diff --git a/server/src/services/native-runtime/status-arbiter.ts b/server/src/services/native-runtime/status-arbiter.ts index e50ff7b66ed..9d3f34ff6b7 100644 --- a/server/src/services/native-runtime/status-arbiter.ts +++ b/server/src/services/native-runtime/status-arbiter.ts @@ -154,6 +154,8 @@ export function arbitrateNativeStatus(input: { boardResponseWaitOrigin?: boolean; isConversation?: boolean; hasActivePauseHold?: boolean; + /** The run only compacted the agent's context; it carries no task outcome. */ + contextCompactionRun?: boolean; reviewOwnerUserId?: string | null; /** Review decisions own task state; a reviewer's finish report cannot override them. */ nativeReviewOutcome?: "resolved" | "pending" | "stale"; @@ -165,6 +167,16 @@ export function arbitrateNativeStatus(input: { agentId: string; priorIssueStatus: NativeAuthoritativeIssueStatus; }): NativeStatusDecision { + if (input.contextCompactionRun) { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "preserve", + toStatus: input.priorIssueStatus, + reasonCode: "context_compaction_status_preserved", + unblockDescriptor: null, + effects: [], + }; + } if (["done", "cancelled"].includes(input.priorIssueStatus)) { return { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index f4697c129f9..87fca0f9de0 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1,4 +1,5 @@ import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; +import { isContextCompactionRun } from "../context-compaction.js"; import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; @@ -3791,6 +3792,7 @@ export function recoveryService( activeExecution: state.hasActiveExecutionPath || latest?.id !== (dispatchRunId ?? run.id), ownedLifecycle: run.issueCommentStatus === "retry_queued" || run.issueCommentStatus === "retry_exhausted" || Boolean(readNonEmptyString(context.goalControlRequestId)) || context.resumeSessionGoalHeartbeat === true || + isContextCompactionRun(context) || isPluginManagedIssueLifecycle(issue) || routine.length > 0 || workspaceChildren.length > 0 || Boolean(goal[0]?.status && goal[0].status !== "complete") || Boolean(active && !ownsRepair), conversation: Boolean(issue.conversationAgentId) || isWaitingConversation(issue), diff --git a/ui/src/api/issues.ts b/ui/src/api/issues.ts index 64f70cef752..49e77870fb7 100644 --- a/ui/src/api/issues.ts +++ b/ui/src/api/issues.ts @@ -372,6 +372,8 @@ export const issuesApi = { ? api.get(`/issues/${id}`, options) : api.get(`/issues/${id}`), getContextUsage: (id: string) => api.get(`/issues/${id}/context-usage`), + compactContext: (id: string) => + api.post<{ requestId: string; runId: string }>(`/issues/${id}/context/compact`, {}), getRunnerGoal: (id: string, agentId?: string | null) => { const query = agentId ? `?agentId=${encodeURIComponent(agentId)}` : ""; return api.get(`/issues/${id}/runner-goal${query}`); diff --git a/ui/src/components/ContextUsageMeter.tsx b/ui/src/components/ContextUsageMeter.tsx index f2b1298d768..954a644cf58 100644 --- a/ui/src/components/ContextUsageMeter.tsx +++ b/ui/src/components/ContextUsageMeter.tsx @@ -1,5 +1,7 @@ import { useRef, useState } from "react"; +import { FoldVertical, Loader2 } from "lucide-react"; import { cn } from "@/lib/utils"; +import { Button } from "@/components/ui/button"; import { timeAgo } from "@/lib/timeAgo"; import { formatContextTokens } from "@/lib/context-tokens"; import { Popover, PopoverContent, PopoverTrigger } from "@/components/ui/popover"; @@ -77,7 +79,44 @@ function Row({ label, value, muted }: { label: string; value: string; muted?: bo ); } -function ContextUsageCard({ usage }: { usage: ContextUsage }) { +export type ContextCompactState = + /** The agent's session can be compacted now. */ + | { status: "available"; onCompact: () => void } + /** A compaction was requested and has not finished yet. */ + | { status: "pending" } + /** Shown disabled with the reason, e.g. the agent is mid-run. */ + | { status: "unavailable"; reason: string }; + +function CompactAction({ compact, level }: { compact: ContextCompactState; level: ContextUsageLevel }) { + return ( +
+ +

+ {compact.status === "unavailable" + ? compact.reason + : compact.status === "pending" + ? "The agent is summarizing its conversation. This usually takes under a minute." + : "Summarizes the conversation so far so the agent has room to keep working."} +

+
+ ); +} + +/** The hover card body; exported for Storybook galleries. */ +export function ContextUsageCard({ usage, compact }: { usage: ContextUsage; compact?: ContextCompactState | null }) { const percent = Math.min(100, Math.round((usage.usedTokens / usage.contextWindow) * 100)); const level = contextUsageLevel(usage); const totals = usage.totals; @@ -124,19 +163,23 @@ function ContextUsageCard({ usage }: { usage: ContextUsage }) { {usage.compactions ? : null} ) : null} + {compact ? : null} ); } /** * Context row value for the task Properties panel: a thin fill bar and - * "64% of 272k". Hover or tap opens the token breakdown. + * "64% of 272k". Hover or tap opens the token breakdown and, when the + * agent's harness supports it, Compact now. */ export function ContextUsageMeter({ usage, + compact, defaultOpen = false, }: { usage: ContextUsage; + compact?: ContextCompactState | null; defaultOpen?: boolean; }) { const [open, setOpen] = useState(defaultOpen); @@ -180,7 +223,7 @@ export function ContextUsageMeter({ onMouseEnter={show} onMouseLeave={hide} > - + ); diff --git a/ui/src/components/IssueProperties.test.tsx b/ui/src/components/IssueProperties.test.tsx index 9588866537d..b56d693a024 100644 --- a/ui/src/components/IssueProperties.test.tsx +++ b/ui/src/components/IssueProperties.test.tsx @@ -114,6 +114,7 @@ vi.mock("../api/instanceSettings", () => ({ vi.mock("../context/ToastContext", () => ({ useToastActions: () => ({ pushToast: vi.fn() }), + useOptionalToastActions: () => ({ pushToast: vi.fn() }), })); vi.mock("../hooks/useProjectOrder", () => ({ diff --git a/ui/src/components/TaskChatThread.tsx b/ui/src/components/TaskChatThread.tsx index b626f43977f..317fc7f4b9f 100644 --- a/ui/src/components/TaskChatThread.tsx +++ b/ui/src/components/TaskChatThread.tsx @@ -66,6 +66,7 @@ import { paperclipRunnerAcceptedResponseWake, paperclipRunnerTimelineItems, prependIssueBrief, + settledCompactionMarkers, settledRunChildren, splitTranscriptAtAnchors, transcriptToTaskChatItems, @@ -1710,6 +1711,10 @@ export function TaskChatThread(props: TaskChatThreadProps) { continue; } previousExecutionWaitKey = null; + const compactionMarkers = settledCompactionMarkers(entries, source.id); + for (const marker of compactionMarkers) { + entriesWithFailures.push({ id: marker.id, ms: toMs(marker.createdAtIso), order: 3, item: marker }); + } const acceptedSummary = acceptedSemanticResultSummary(meta?.resultJson); const parsedSource = transcriptToTaskChatItems(entries, { runId: source.id, @@ -2160,8 +2165,11 @@ export function TaskChatThread(props: TaskChatThreadProps) { rest: partitionRunHealthItems(timelineItemsBySegment[segmentIndex] ?? []).rest, }; const children = settledRunChildren(runHealth.rest); - if (children.length === 0 && !finalResponse && !sourceIsPaperclipRunner) + // A compaction-only run is fully shown by its divider. + if (children.length === 0 && !finalResponse && (!sourceIsPaperclipRunner || compactionMarkers.length > 0)) { + if (compactionMarkers.length > 0) settledRunIds.add(source.id); continue; + } settledRunIds.add(source.id); if (finalResponse) settledReplyRunIds.add(source.id); const segmented = timelineAnchors.length > 0; diff --git a/ui/src/components/issue-properties/IssueProperties.tsx b/ui/src/components/issue-properties/IssueProperties.tsx index 848d447c317..b15987f2de2 100644 --- a/ui/src/components/issue-properties/IssueProperties.tsx +++ b/ui/src/components/issue-properties/IssueProperties.tsx @@ -2769,7 +2769,7 @@ export function IssueProperties({ {contextUsage !== undefined ? contextUsage ? {contextUsage} : null - : } + : } {workspacePickerEligible || hasWorkspaceRuntimeControls || issue.currentExecutionWorkspace?.branchName || issue.currentExecutionWorkspace?.cwd || issue.executionWorkspaceId ? ( diff --git a/ui/src/components/issue-properties/context-usage-row.test.tsx b/ui/src/components/issue-properties/context-usage-row.test.tsx index cc1d100e1de..8dc08ade4c5 100644 --- a/ui/src/components/issue-properties/context-usage-row.test.tsx +++ b/ui/src/components/issue-properties/context-usage-row.test.tsx @@ -3,8 +3,8 @@ import { flushSync } from "react-dom"; import { createRoot, type Root } from "react-dom/client"; import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -const api = vi.hoisted(() => ({ getContextUsage: vi.fn() })); -vi.mock("@/api/issues", () => ({ issuesApi: { getContextUsage: api.getContextUsage } })); +const api = vi.hoisted(() => ({ getContextUsage: vi.fn(), compactContext: vi.fn() })); +vi.mock("@/api/issues", () => ({ issuesApi: { getContextUsage: api.getContextUsage, compactContext: api.compactContext } })); import { ApiError } from "@/api/api-error"; import { IssueContextUsageRow } from "./context-usage-row"; @@ -28,10 +28,10 @@ afterEach(() => { document.body.innerHTML = ""; }); -function render(assigneeAgentId: string | null = "agent-1") { +function render(assigneeAgentId: string | null = "agent-1", issueStatus = "in_progress") { flushSync(() => root.render( - + , )); } @@ -75,6 +75,54 @@ describe("IssueContextUsageRow", () => { expect(container.textContent).toBe(""); }); + function openCard() { + const trigger = container.querySelector("[data-testid=context-usage-meter]"); + expect(trigger).toBeTruthy(); + flushSync(() => trigger!.click()); + return document.body.querySelector("[data-testid=context-usage-compact]"); + } + + it("asks the agent to compact from the hover card", async () => { + api.getContextUsage.mockResolvedValue({ ...reading, compaction: { status: "available", reasonCode: null, reason: null } }); + api.compactContext.mockReturnValue(new Promise(() => {})); + render(); await settle(); + const button = openCard(); + expect(button?.textContent).toContain("Compact now"); + expect(button?.disabled).toBe(false); + flushSync(() => button!.click()); await settle(); + expect(api.compactContext).toHaveBeenCalledWith("issue-1"); + expect(document.body.querySelector("[data-testid=context-usage-compact]")?.textContent).toContain("Compacting…"); + }); + + it("explains why compaction is unavailable", async () => { + api.getContextUsage.mockResolvedValue({ + ...reading, + compaction: { status: "unavailable", reasonCode: "agent_busy", reason: "Available when the agent finishes its current run." }, + }); + render(); await settle(); + const button = openCard(); + expect(button?.disabled).toBe(true); + expect(document.body.textContent).toContain("Available when the agent finishes its current run."); + }); + + it("refreshes Compact now when the task status changes", async () => { + api.getContextUsage + .mockResolvedValueOnce({ ...reading, compaction: { status: "available", reasonCode: null, reason: null } }) + .mockResolvedValue({ ...reading, compaction: { status: "unavailable", reasonCode: "task_closed", reason: "Available while the task is open." } }); + render("agent-1", "in_progress"); await settle(); + render("agent-1", "done"); await settle(); + expect(api.getContextUsage).toHaveBeenCalledTimes(2); + expect(openCard()?.disabled).toBe(true); + }); + + it("shows a compaction already in progress", async () => { + api.getContextUsage.mockResolvedValue({ ...reading, compaction: { status: "pending", reasonCode: null, reason: null, runId: "run-2" } }); + render(); await settle(); + const button = openCard(); + expect(button?.disabled).toBe(true); + expect(button?.textContent).toContain("Compacting…"); + }); + it("keeps the last reading when a refresh fails", async () => { api.getContextUsage.mockResolvedValueOnce(reading).mockRejectedValue(new ApiError("Internal error", 500, { error: "internal" })); render(); await settle(); diff --git a/ui/src/components/issue-properties/context-usage-row.tsx b/ui/src/components/issue-properties/context-usage-row.tsx index 4cf177c0561..2da8f622933 100644 --- a/ui/src/components/issue-properties/context-usage-row.tsx +++ b/ui/src/components/issue-properties/context-usage-row.tsx @@ -1,10 +1,11 @@ import { useEffect, useRef } from "react"; -import { useQuery } from "@tanstack/react-query"; -import type { IssueContextUsage } from "@paperclipai/shared"; +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import type { IssueContextCompaction, IssueContextUsage } from "@paperclipai/shared"; import { issuesApi } from "../../api/issues"; import { queryKeys } from "../../lib/queryKeys"; +import { useOptionalToastActions } from "../../context/ToastContext"; import { QueryErrorState, useQueryView } from "../QueryView"; -import { ContextUsageMeter, type ContextUsage } from "../ContextUsageMeter"; +import { ContextUsageMeter, type ContextCompactState, type ContextUsage } from "../ContextUsageMeter"; import { PropertyRow } from "./primitives"; /** Refresh cadence while an agent run is reporting usage. */ @@ -27,6 +28,20 @@ export function contextUsageFromIssue(value: IssueContextUsage): ContextUsage { }; } +/** Maps the server's Compact now state, with a click in flight shown as pending. */ +export function contextCompactState( + compaction: IssueContextCompaction | undefined, + requesting: boolean, + onCompact: () => void, +): ContextCompactState | null { + if (!compaction) return null; + if (requesting || compaction.status === "pending") return { status: "pending" }; + if (compaction.status === "unavailable") { + return { status: "unavailable", reason: compaction.reason ?? "Compaction is unavailable right now." }; + } + return { status: "available", onCompact }; +} + /** * Context row for the task Properties panel. Renders nothing until the * agent's harness has reported a context window for this task. A cached @@ -36,11 +51,14 @@ export function contextUsageFromIssue(value: IssueContextUsage): ContextUsage { export function IssueContextUsageRow({ issueId, assigneeAgentId, + issueStatus, hasActiveRun, }: { issueId: string; /** The reading belongs to the assignee's session, so a reassignment refetches it. */ assigneeAgentId: string | null; + /** Compact now depends on the task being open, so a status change refreshes it. */ + issueStatus?: string; hasActiveRun: boolean; }) { const query = useQuery({ @@ -50,6 +68,27 @@ export function IssueContextUsageRow({ staleTime: hasActiveRun ? 0 : 60_000, }); const view = useQueryView(query); + const queryClient = useQueryClient(); + const toast = useOptionalToastActions(); + const compact = useMutation({ + mutationFn: () => issuesApi.compactContext(issueId), + onSettled: async () => { + // The compaction runs as a short agent run; refresh the run state that + // drives polling, then the reading itself. + await Promise.all([ + queryClient.invalidateQueries({ queryKey: queryKeys.issues.activeRun(issueId) }), + queryClient.invalidateQueries({ queryKey: queryKeys.issues.liveRuns(issueId) }), + queryClient.invalidateQueries({ queryKey: queryKeys.issues.contextUsage(issueId, assigneeAgentId) }), + ]); + }, + onError: (error) => { + toast?.pushToast({ + title: "Could not compact the context", + body: error instanceof Error ? error.message : "Try again in a moment.", + tone: "error", + }); + }, + }); // Pick up the final measurement once the run settles. const wasActive = useRef(hasActiveRun); const { refetch } = query; @@ -57,11 +96,19 @@ export function IssueContextUsageRow({ if (wasActive.current && !hasActiveRun) void refetch(); wasActive.current = hasActiveRun; }, [hasActiveRun, refetch]); + const previousStatus = useRef(issueStatus); + useEffect(() => { + if (previousStatus.current !== issueStatus) void refetch(); + previousStatus.current = issueStatus; + }, [issueStatus, refetch]); if (view.data) { return ( - + compact.mutate())} + /> ); } diff --git a/ui/src/components/task-chat/transcript-adapter.ts b/ui/src/components/task-chat/transcript-adapter.ts index bfe6d702a64..de43bfdc1ac 100644 --- a/ui/src/components/task-chat/transcript-adapter.ts +++ b/ui/src/components/task-chat/transcript-adapter.ts @@ -1284,7 +1284,26 @@ export function transcriptToTaskChatItems( export function settledRunChildren( parsed: readonly TaskChatItem[], ): TaskChatTurnChildItem[] { - return buildTurnTimelineRows(parsed, false); + // A finished run's compactions render as thread dividers (see + // settledCompactionMarkers), not inside its collapsed turn. + return buildTurnTimelineRows(parsed, false).filter((item) => !isCompactionDivider(item)); +} + +/** + * Compaction dividers of a finished run. A compaction divides the whole + * conversation, so the thread places them between messages rather than + * inside the run's collapsed activity, where they would stay hidden. + */ +export function settledCompactionMarkers( + entries: readonly TranscriptEntry[], + runId: string, +): TaskChatMarkerItem[] { + const markers: TaskChatMarkerItem[] = []; + entries.forEach((entry, index) => { + if (entry.kind !== "provider_activity" || entry.eventType !== "context.compacted") return; + markers.push(contextCompactedMarker(entry, `${runId}:compaction:${String(entry.payload.compactionId ?? "")}:${index}`)); + }); + return markers; } /** diff --git a/ui/storybook/stories/tasks/task-context-window.stories.tsx b/ui/storybook/stories/tasks/task-context-window.stories.tsx index 45aacbda9a9..9874fce56c1 100644 --- a/ui/storybook/stories/tasks/task-context-window.stories.tsx +++ b/ui/storybook/stories/tasks/task-context-window.stories.tsx @@ -4,7 +4,7 @@ import type { Meta, StoryObj } from "@storybook/react-vite"; import { TaskChatThreadView } from "@/components/task-chat/TaskChatThreadView"; import { TaskChatMarker } from "@/components/task-chat/TaskChatMarker"; import { TaskChatPresentationProvider } from "@/components/task-chat/presentation-mode"; -import { ContextUsageMeter, type ContextUsage } from "@/components/ContextUsageMeter"; +import { ContextUsageCard, ContextUsageMeter, type ContextCompactState, type ContextUsage } from "@/components/ContextUsageMeter"; import { IssueProperties } from "@/components/IssueProperties"; import { ScrollArea } from "@/components/ui/scroll-area"; import { queryKeys } from "@/lib/queryKeys"; @@ -165,7 +165,41 @@ function StorybookData({ children }: { children: ReactNode }) { return ready ? children : null; } -function PropertiesPane({ usage, defaultOpen }: { usage: ContextUsage; defaultOpen?: boolean }) { +const COMPACT_AVAILABLE: ContextCompactState = { status: "available", onCompact: () => {} }; +const COMPACT_PENDING: ContextCompactState = { status: "pending" }; +const COMPACT_MID_RUN: ContextCompactState = { + status: "unavailable", + reason: "Available when the agent finishes its current run.", +}; +const COMPACT_UNSUPPORTED: ContextCompactState = { + status: "unavailable", + reason: "This agent's harness can't compact on request.", +}; + +/** Compact now flips to pending, then the usage drops and a compaction is counted. */ +function useCompactDemo(initial: ContextUsage) { + const [usage, setUsage] = useState(initial); + const [pending, setPending] = useState(false); + const compact: ContextCompactState = pending + ? COMPACT_PENDING + : { + status: "available", + onCompact: () => { + setPending(true); + setTimeout(() => { + setPending(false); + setUsage((current) => ({ + ...current, + usedTokens: Math.round(current.usedTokens * 0.17), + compactions: (current.compactions ?? 0) + 1, + })); + }, 1800); + }, + }; + return { usage, compact }; +} + +function PropertiesPane({ usage, compact, defaultOpen }: { usage: ContextUsage; compact?: ContextCompactState | null; defaultOpen?: boolean }) { return (
@@ -176,7 +210,7 @@ function PropertiesPane({ usage, defaultOpen }: { usage: ContextUsage; defaultOp issue={issue} childIssues={[]} onUpdate={() => undefined} - contextUsage={} + contextUsage={} />
@@ -206,16 +240,26 @@ function StateCard({ title, rule, children }: { title: string; rule: string; chi } /** A Properties-style row without the full panel, for comparing states. */ -function MeterRow({ usage }: { usage: ContextUsage }) { +function MeterRow({ usage, compact }: { usage: ContextUsage; compact?: ContextCompactState | null }) { return (
Context - + +
+ ); +} + +/** The hover card as it renders in the popover, laid out inline for comparison. */ +function HoverCard({ usage, compact }: { usage: ContextUsage; compact: ContextCompactState }) { + return ( +
+
); } function TaskPage() { + const { usage, compact } = useCompactDemo(USAGE.nearCompaction); return (
@@ -225,7 +269,7 @@ function TaskPage() {
- +
); @@ -239,22 +283,34 @@ function AllStates() {

Context window

A Context row in the task's Properties shows how full the assigned agent's context window is. Hover - or tap it for the token breakdown. Codex reports usage through{" "} + or tap it for the token breakdown and Compact now. Codex reports usage through{" "} thread/tokenUsage/updated, Claude through ACP{" "} usage_update.

- + - + - + - + + + + + + + + + + + + +

(no row)

@@ -271,25 +327,30 @@ function AllStates() { ); } -function Breakdown({ usage }: { usage: ContextUsage }) { +function Breakdown({ usage, compact }: { usage: ContextUsage; compact?: ContextCompactState | null }) { return (
- +
); } +function InteractiveBreakdown() { + const { usage, compact } = useCompactDemo(USAGE.nearCompaction); + return ; +} + const meta = { id: "product-task-context-window", title: "Tasks and conversations/Task creation and lifecycle/Product/Task chat/Context window", component: ContextUsageMeter, - args: { usage: USAGE.working }, + args: { usage: USAGE.working, compact: COMPACT_AVAILABLE }, parameters: { layout: "fullscreen", docs: { description: { component: - "Context row in the task Properties panel with a hover breakdown, plus the transcript divider recorded when the agent's context is compacted.", + "Context row in the task Properties panel with a hover breakdown and Compact now, plus the transcript divider recorded when the agent's context is compacted.", }, }, paperclip: { @@ -299,7 +360,7 @@ const meta = { ], "implementation": "production", "boundary": "fixtures", - "notes": "Renders the context meter inside the production Issue Properties panel and the compaction divider in the production thread view, from fixture usage snapshots." + "notes": "Renders the context meter inside the production Issue Properties panel and the compaction divider in the production thread view, from fixture usage snapshots. Compact now is simulated locally; no agent session is contacted." }, }, render: (args) => ( @@ -313,6 +374,8 @@ type Story = StoryObj; export const InTask: Story = { render: () => }; export const AllStatesGallery: Story = { render: () => }; -export const BreakdownNearCompaction: Story = { render: () => }; -export const BreakdownWorking: Story = { render: () => }; -export const BreakdownClaudeWithCost: Story = { render: () => }; +export const BreakdownNearCompaction: Story = { render: () => }; +export const BreakdownCompacting: Story = { render: () => }; +export const BreakdownMidRun: Story = { render: () => }; +export const BreakdownWorking: Story = { render: () => }; +export const BreakdownClaudeWithCost: Story = { render: () => };