From 4e88f0fe4cca9009f6307aec7bea4903944bad0d Mon Sep 17 00:00:00 2001 From: James Maes Date: Tue, 12 May 2026 17:39:25 -0500 Subject: [PATCH 1/2] =?UTF-8?q?fix(mcp):=20remove=20redundant=20inner=20to?= =?UTF-8?q?ol-result=20cap=20=E2=80=94=20closes=20#21?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ToolResultPacker.pack was capping bytes that had already been capped at the MCP dispatcher boundary by SanitizeForModel.prepareForBoundary, making the doc-comment lie ("full blob still flows to ReplayLog ... observability is not lossy") provably wrong. The orchestrator's .toolResultFull replay row carried the already-capped 8 KB bytes, not the original helper output, and the second cap was redundant. Fix: turn ToolResultPacker.pack into a pass-through decoder. The MCP dispatcher's prepareForBoundary call (sanitize + 8 KB head-truncate with the …[tool-result-truncated at 8192 bytes] marker) is the sole cap site. ReplayLog rows now carry exactly the dispatcher-prepared bytes — never less, never re-marked. The packer's API stability is preserved (Packed.{modelFacing, fullBytes, wasCapped, omittedByteCount} all still present); wasCapped reports informationally from the size signal and omittedByteCount is always zero post-fix. Doc-comment updated to describe the new contract truthfully. Tests updated: - ToolResultPackerTests: TP2 / TP5 renamed to assert the packer no longer adds the [TRUNCATED: ...] marker (the dispatcher's …[tool-result-truncated at 8192 bytes] marker is the only one). - OrchestratorSubmitTests: - OS6 renamed to test_OS6_replayAndModelSeeDispatcherPreparedBytes, asserting both replay and the model see the dispatcher's prepared bytes (no double-cap, no double-marker). - OS6b is the new #21 regression: a dispatcher that returns exactly 8 KB drives a persisted ReplayLog row of exactly 8 KB (verified by opening the SQLite file directly) — not 4 KB or less. Refs audit-2026-05-12 CRIT-2 (.planning/audit-2026-05-12/mcp-agent-audit.md). Co-Authored-By: Claude Opus 4.7 --- .../Sources/AgentCore/ToolResultPacker.swift | 74 +++++++----- .../ToolResultPackerTests.swift | 80 +++++++------ .../OrchestratorSubmitTests.swift | 109 +++++++++++++++--- 3 files changed, 187 insertions(+), 76 deletions(-) diff --git a/packages/AgentCore/Sources/AgentCore/ToolResultPacker.swift b/packages/AgentCore/Sources/AgentCore/ToolResultPacker.swift index 7679327..0579f0b 100644 --- a/packages/AgentCore/Sources/AgentCore/ToolResultPacker.swift +++ b/packages/AgentCore/Sources/AgentCore/ToolResultPacker.swift @@ -1,6 +1,7 @@ import Foundation -/// AGENT-08 8 KB tool-result cap. +/// AGENT-08 8 KB tool-result cap — wire-up between MCP boundary defense and +/// the LLM message history. /// /// **Why 8 KB?** Opus 4.7 tokenizer produces ~1.35× tokens vs Opus 3.x; /// 8 KB at the new ratio ≈ 12 K tokens. Three tool calls / turn × 12 K @@ -8,52 +9,67 @@ import Foundation /// against context blowout. Revisit empirically via DevOverlay's /// cache-ratio signal in P5+. /// -/// The cap is **model-facing only**. The full blob still flows to -/// `ReplayLog` via `ReplayEvent.toolResultFull(toolUseId:bytes:)` — -/// observability is not lossy (OBS-02). +/// **Where the cap actually fires:** at the MCP dispatcher boundary via +/// `SanitizeForModel.prepareForBoundary(_, capBytes: 8192)` (lives in the +/// JarvisMCP package). By the time bytes reach this packer, they are +/// already sanitized + head-truncated to ≤ 8 KB. `pack(_:)` does NOT cap +/// again — that was the CRIT-2 audit-2026-05-12 double-cap, fixed by +/// closing #21. The packer's job today is reduced to: decode the +/// already-prepared bytes into a `String` for the LLM message history and +/// pass the same bytes through to `ReplayLog` via +/// `ReplayEvent.toolResultFull(toolUseId:bytes:)`. The audit-trail row +/// reflects the boundary-prepared bytes — not the original helper output +/// (per the audit's stated v0.1 acceptance: replay row = 8 KB cap, not the +/// full original blob). public enum ToolResultPacker { - /// Cap for model-facing tool_result content in bytes. + /// Cap for model-facing tool_result content in bytes. Kept for the + /// (informational) `wasCapped` / `omittedByteCount` fields and to + /// document the contract; the cap itself is enforced at the MCP + /// dispatcher boundary (`SanitizeForModel.prepareForBoundary`). public static let modelFacingCapBytes: Int = 8 * 1024 /// Result of packing a single tool result. public struct Packed: Sendable { /// String the orchestrator embeds in the next `LLMMessage` of role - /// `.tool`. UTF-8 decoded from the (possibly capped) byte prefix; if - /// truncated, ends with `\n\n[TRUNCATED: …]` marker text. + /// `.tool`. UTF-8 decoded from `fullBytes` as-is — the MCP + /// dispatcher boundary already applied the 8 KB head-truncate + + /// the `…[tool-result-truncated at 8192 bytes]` marker. public let modelFacing: String - /// Original bytes — destined for `ReplayEvent.toolResultFull`. + /// The dispatcher-prepared bytes — destined for + /// `ReplayEvent.toolResultFull`. After #21 these are the same + /// bytes the model sees (no double-cap). public let fullBytes: Data - /// True if `fullBytes.count > modelFacingCapBytes`. + /// True if the dispatcher truncated. Detected by looking at + /// the input size relative to `modelFacingCapBytes` — informational + /// only since `pack(_:)` no longer truncates. public let wasCapped: Bool - /// Number of bytes omitted from `modelFacing`. Zero when not capped. + /// Always zero after #21 — kept for API stability. The dispatcher + /// strips bytes before the packer sees them; the omitted count is + /// not knowable here. public let omittedByteCount: Int } - /// Pack one tool result. The caller (`AgentOrchestrator`) routes the - /// `modelFacing` string through `UntrustedWrapper` before appending to - /// the LLM message history, and emits `fullBytes` to `ReplayLog`. + /// Pack one tool result. Input bytes have already been sanitized + + /// 8-KB-head-truncated by `SanitizeForModel.prepareForBoundary` at the + /// MCP dispatcher boundary; this packer does NOT cap again (per #21 + /// closure). The caller (`AgentOrchestrator`) routes `modelFacing` + /// through `UntrustedWrapper` before appending to the LLM message + /// history and emits `fullBytes` to `ReplayLog`. public static func pack(_ raw: Data) -> Packed { - let cap = modelFacingCapBytes - let capped = raw.prefix(cap) - let wasCapped = raw.count > capped.count - let omitted = raw.count - capped.count - let modelFacing: String - if wasCapped { - // String(decoding:as:) replaces invalid bytes with U+FFFD; we may - // truncate mid-codepoint at the byte boundary, that's acceptable. - let head = String(decoding: capped, as: UTF8.self) - modelFacing = head + "\n\n[TRUNCATED: \(omitted) bytes omitted; full blob in replay log]" - } else { - modelFacing = String(decoding: capped, as: UTF8.self) - } + // The boundary cap fires upstream in + // `SanitizeForModel.prepareForBoundary(_:capBytes:)`. We just + // decode + pass through. `wasCapped` is reported informationally + // from the size signal so callers / dashboards that previously + // read it keep working; `omittedByteCount` is no longer knowable + // here and is always zero. return Packed( - modelFacing: modelFacing, + modelFacing: String(decoding: raw, as: UTF8.self), fullBytes: raw, - wasCapped: wasCapped, - omittedByteCount: omitted + wasCapped: raw.count >= modelFacingCapBytes, + omittedByteCount: 0 ) } } diff --git a/packages/AgentCore/Tests/AgentCoreTests/ToolResultPackerTests.swift b/packages/AgentCore/Tests/AgentCoreTests/ToolResultPackerTests.swift index 3630464..cb86c0c 100644 --- a/packages/AgentCore/Tests/AgentCoreTests/ToolResultPackerTests.swift +++ b/packages/AgentCore/Tests/AgentCoreTests/ToolResultPackerTests.swift @@ -1,8 +1,16 @@ import XCTest @testable import AgentCore +// Tests reflect the #21 (audit-2026-05-12 CRIT-2) closure: `ToolResultPacker` +// no longer truncates. The 8 KB cap fires at the MCP dispatcher boundary via +// `SanitizeForModel.prepareForBoundary`; the packer is a pass-through that +// decodes the dispatcher-prepared bytes for the LLM message history and +// hands the same bytes to ReplayLog. `wasCapped` is informational only +// (derived from input size); `omittedByteCount` is always zero since the +// dispatcher already stripped the omitted bytes before the packer sees them. final class ToolResultPackerTests: XCTestCase { - /// TP1: 4 KB input → no truncation; full string returned; full bytes preserved. + + /// TP1: under-cap input flows through unchanged; no truncation marker. func test_TP1_smallInputNotCapped() { let raw = Data(repeating: UInt8(ascii: "a"), count: 4 * 1024) let packed = ToolResultPacker.pack(raw) @@ -13,54 +21,60 @@ final class ToolResultPackerTests: XCTestCase { XCTAssertFalse(packed.modelFacing.contains("[TRUNCATED")) } - /// TP2: 16 KB input → first 8192 bytes + truncation marker; full bytes preserved. - func test_TP2_largeInputCappedWithMarker() { + /// TP2 (post-#21): packer does NOT truncate. The dispatcher already + /// truncated upstream — bytes flowing in are ≤ 8 KB. The packer passes + /// them through unchanged, and `fullBytes == raw` regardless of size. + func test_TP2_packerDoesNotTruncate() { + // Simulate the rare/buggy case where a caller hands us larger bytes + // (the production path won't, but the API stability matters): packer + // still doesn't truncate. let raw = Data(repeating: UInt8(ascii: "x"), count: 16 * 1024) let packed = ToolResultPacker.pack(raw) - XCTAssertTrue(packed.wasCapped) - XCTAssertEqual(packed.omittedByteCount, 16 * 1024 - 8192) - XCTAssertEqual(packed.fullBytes.count, 16 * 1024) - XCTAssertTrue(packed.modelFacing.hasPrefix(String(repeating: "x", count: 8192)), - "modelFacing must begin with the first 8192 bytes") - XCTAssertTrue(packed.modelFacing.contains("[TRUNCATED:"), - "must contain truncation marker") - XCTAssertTrue(packed.modelFacing.contains("full blob in replay log"), - "must reference replay log") + // After #21 the packer never inserts a TRUNCATED marker; the + // dispatcher's `SanitizeForModel.headTruncate` is the only marker + // source. + XCTAssertFalse(packed.modelFacing.contains("[TRUNCATED:")) + XCTAssertEqual(packed.fullBytes.count, 16 * 1024, + "fullBytes is a pass-through of the input") + XCTAssertEqual(packed.modelFacing.count, 16 * 1024, + "modelFacing is a pass-through decoding of the input") + XCTAssertEqual(packed.omittedByteCount, 0, + "packer no longer knows omitted-count; always zero") } - /// TP3: exactly 8192 bytes — equal-to-cap is NOT over. - func test_TP3_exactlyAtCapNotTruncated() { + /// TP3: input exactly at cap reports `wasCapped == true` (size >= cap) + /// but no marker is added — the packer is purely informational. + func test_TP3_exactlyAtCapInformational() { let raw = Data(repeating: UInt8(ascii: "y"), count: 8192) let packed = ToolResultPacker.pack(raw) - XCTAssertFalse(packed.wasCapped) + XCTAssertTrue(packed.wasCapped, + "size >= cap reports true informationally") XCTAssertEqual(packed.omittedByteCount, 0) XCTAssertFalse(packed.modelFacing.contains("[TRUNCATED")) } - /// TP4: multi-byte UTF-8 boundary at byte 8192 — String(decoding:as:) replaces - /// invalid bytes with U+FFFD and does not crash. - func test_TP4_utf8BoundaryDoesNotCrash() { - // Build a buffer of size 16K where bytes 8190..8192 sit on a 4-byte - // emoji boundary. emoji "😀" is F0 9F 98 80; place at offset 8190 so - // bytes 8190,8191 are F0 9F (start of emoji) and the cut at 8192 cuts - // mid-codepoint. - var raw = [UInt8](repeating: UInt8(ascii: "z"), count: 16 * 1024) + /// TP4: UTF-8 decode of input never crashes — `String(decoding:as:)` + /// replaces invalid sequences with U+FFFD. Production input has been + /// scrubbed by `SanitizeForModel.sanitize` upstream so invalid sequences + /// are vanishingly rare, but the API must not crash on adversarial input. + func test_TP4_utf8DecodeNeverCrashes() { + // Bytes 8190..8191 are the start of a 4-byte emoji — incomplete by + // themselves. `String(decoding:as:)` substitutes U+FFFD; no crash. + var bytes = [UInt8](repeating: UInt8(ascii: "z"), count: 16 * 1024) let emoji: [UInt8] = [0xF0, 0x9F, 0x98, 0x80] for (i, b) in emoji.enumerated() { - raw[8190 + i] = b + bytes[8190 + i] = b } - let data = Data(raw) - // Should not crash even though bytes 8190..8191 form an incomplete prefix. - let packed = ToolResultPacker.pack(data) - XCTAssertTrue(packed.wasCapped) - XCTAssertNotNil(packed.modelFacing) + let packed = ToolResultPacker.pack(Data(bytes)) + XCTAssertFalse(packed.modelFacing.isEmpty) } - /// TP5: truncation marker is grep-able and deterministic. - func test_TP5_truncationMarkerIsGreppable() { + /// TP5 (post-#21): the packer never adds the `[TRUNCATED: …]` marker. + /// The dispatcher's `…[tool-result-truncated at 8192 bytes]` marker is + /// the only one a downstream observer should ever see. + func test_TP5_packerNeverAddsTruncatedMarker() { let raw = Data(repeating: UInt8(ascii: "a"), count: 9000) let packed = ToolResultPacker.pack(raw) - XCTAssertTrue(packed.modelFacing.contains("TRUNCATED")) - XCTAssertTrue(packed.modelFacing.contains("full blob in replay log")) + XCTAssertFalse(packed.modelFacing.contains("[TRUNCATED:")) } } diff --git a/packages/AgentCore/Tests/AgentOrchestratorTests/OrchestratorSubmitTests.swift b/packages/AgentCore/Tests/AgentOrchestratorTests/OrchestratorSubmitTests.swift index 43c2069..5f6be97 100644 --- a/packages/AgentCore/Tests/AgentOrchestratorTests/OrchestratorSubmitTests.swift +++ b/packages/AgentCore/Tests/AgentOrchestratorTests/OrchestratorSubmitTests.swift @@ -234,14 +234,22 @@ final class OrchestratorSubmitTests: XCTestCase { // MARK: - OS6 - /// OS6: ReplayLog gets a `.toolResultFull` with the FULL bytes (not the - /// 8KB-capped model-facing string). We assert this by sending a 16 KB blob - /// from the dispatcher and checking the model-facing message in the second - /// provider call has the truncation marker (proving cap was applied) AND - /// the full bytes are still preserved in `Packed.fullBytes`. - func test_OS6_replayGetsFullBlobModelGetsCapped() async throws { + /// OS6 (post-#21 / audit-2026-05-12 CRIT-2): the 8 KB cap is the MCP + /// dispatcher's responsibility (`SanitizeForModel.prepareForBoundary`). + /// `ToolResultPacker.pack` is now a pass-through. The orchestrator + /// writes the dispatcher's prepared bytes to ReplayLog as + /// `.toolResultFull` and embeds them in the model-facing tool_result. + /// This test simulates a dispatcher that has already applied the cap + /// (returns exactly 8 KB) and asserts both the replay-bound bytes and + /// the model-facing message are sized at the cap — NOT 4 KB (which a + /// remnant per-line cap would produce) and NOT 16 KB (which would + /// indicate the cap was bypassed entirely). + func test_OS6_replayAndModelSeeDispatcherPreparedBytes() async throws { let toolReq = ToolUseRequest(id: "tu1", name: "big", argsJSON: Data("{}".utf8)) - let bigBlob = Data(repeating: UInt8(ascii: "X"), count: 16 * 1024) + // Dispatcher returns exactly 8 KB — matching the contract that the + // MCP boundary applies `prepareForBoundary(_, capBytes: 8192)` + // upstream of the orchestrator. + let preparedBlob = Data(repeating: UInt8(ascii: "X"), count: 8 * 1024) let mock = MockLLMProvider(scripts: [ .init(events: [ @@ -256,7 +264,7 @@ final class OrchestratorSubmitTests: XCTestCase { .messageStop, ]), ]) - let dispatcher = StubToolDispatcher(dispatch: { _ in bigBlob }) + let dispatcher = StubToolDispatcher(dispatch: { _ in preparedBlob }) let (orch, _, _) = try await buildOrchestrator(provider: mock, dispatcher: dispatcher) @@ -271,12 +279,85 @@ final class OrchestratorSubmitTests: XCTestCase { guard case .toolResult(_, let content) = secondCallToolMsg.content[0] else { return XCTFail("expected toolResult content") } - XCTAssertTrue(content.contains("[TRUNCATED:"), - "model-facing tool result must be capped + carry truncation marker") - // Model-facing is wrapped in nonce envelope; total length is wrapper + - // 8192 bytes + marker text. Assert it's far smaller than 16 KB. - XCTAssertLessThan(content.utf8.count, 9000, - "model-facing must be < 9 KB (8 KB cap + small marker + envelope)") + // The packer no longer adds a `[TRUNCATED: …]` marker — the + // dispatcher's `…[tool-result-truncated at 8192 bytes]` marker is + // the only one a downstream consumer should ever see. The + // dispatcher stub in this test does NOT add one (its job is to + // simulate already-prepared bytes), so the model-facing string + // should not contain either marker. + XCTAssertFalse(content.contains("[TRUNCATED:"), + "packer must not double-mark — dispatcher owns the truncation marker") + // The 8 KB cap bytes flow through; the model-facing content is the + // dispatcher payload wrapped in the SEC-06 nonce envelope. Assert + // the payload byte count is preserved (>= 8 KB, accounting for the + // wrapper text; NOT 4 KB or less). + XCTAssertGreaterThanOrEqual(content.utf8.count, 8 * 1024, + "model-facing must carry the full 8 KB dispatcher payload — not 4 KB or less") + } + + /// OS6b (#21 regression — audit-2026-05-12 CRIT-2): the persisted + /// ReplayLog row for `.toolResultFull` is keyed to exactly the bytes + /// the dispatcher returned (no second cap, no per-line strip applied + /// on top). A dispatcher that returns 8 KB → replay row contains + /// exactly 8 KB of payload. Pre-fix, `ToolResultPacker.pack`'s redundant + /// inner cap re-truncated already-capped bytes (no-op for size but the + /// design was wrong); a per-line variant could have dropped to 4 KB + /// under different inputs. This test pins the post-fix invariant by + /// driving end-to-end through the orchestrator and reading the + /// `payload_bytes` BLOB from the SQLite `events` table. + func test_OS6b_replayRowMatchesDispatcherBytesNoSecondCap() async throws { + let toolReq = ToolUseRequest(id: "tu-21", name: "big", argsJSON: Data("{}".utf8)) + // Exactly 8 KB — what a real dispatcher emits after + // `SanitizeForModel.prepareForBoundary(_, capBytes: 8192)`. + let preparedBlob = Data(repeating: UInt8(ascii: "X"), count: 8 * 1024) + + let mock = MockLLMProvider(scripts: [ + .init(events: [ + .messageStart(LLMMessageStart(messageId: "m1", model: "x", usagePrefix: nil)), + .toolUseRequested(toolReq), + .stopReason(.toolUse), + .messageStop, + ]), + .init(events: [ + .messageStart(LLMMessageStart(messageId: "m2", model: "x", usagePrefix: nil)), + .stopReason(.endTurn), + .messageStop, + ]), + ]) + let dispatcher = StubToolDispatcher(dispatch: { _ in preparedBlob }) + + let (orch, replay, _) = try await buildOrchestrator(provider: mock, dispatcher: dispatcher) + + _ = await orch.submit(.text("fetch")) + _ = await collectUntil(orch: orch) { ev in + if case .stateChange(.idle) = ev { return true }; return false + } + + // Flush + close so the row is on disk before we open a competing + // connection. + await replay.flush() + try await replay.close() + + let conn = try SQLiteConnection.open(at: tempHome.dbURL) + defer { try? conn.close() } + let blob: Data? = try conn.query( + "SELECT payload_bytes FROM events WHERE kind='tool_result_full';", + map: { $0.columnBlob(at: 0) } + ).first ?? nil + + XCTAssertNotNil(blob, "must find a tool_result_full row") + guard let blob else { return } + // The envelope is `{"tool_use_id": "...", "bytes": ""}`. + let json = try JSONSerialization.jsonObject(with: blob) as? [String: String] + XCTAssertEqual(json?["tool_use_id"], "tu-21") + let recovered = Data(base64Encoded: json?["bytes"] ?? "") ?? Data() + // #21 invariant: the replay row carries EXACTLY the dispatcher's + // 8 KB payload — not 4 KB (which a residual per-line strip would + // produce) and not less. + XCTAssertEqual(recovered.count, 8 * 1024, + "replay row must be exactly 8 KB — not 4 KB or less (#21)") + XCTAssertEqual(recovered, preparedBlob, + "replay row must be byte-for-byte identical to the dispatcher payload") } // MARK: - OS7 -- 2.54.0 From d73cfc6627cf67bbe14917233fa3f955da5c30f6 Mon Sep 17 00:00:00 2001 From: James Maes Date: Tue, 12 May 2026 17:39:48 -0500 Subject: [PATCH 2/2] =?UTF-8?q?fix(mcp):=20wire=20turnIDResolver=20post-in?= =?UTF-8?q?stallAgent=20=E2=80=94=20closes=20#20?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MCPRuntimeWiring.build constructs ReplayingToolResultObserver with a `{ nil }` placeholder resolver because the AgentOrchestrator that owns the in-flight TurnID doesn't yet exist (the orchestrator consumes the dispatcher chain we return). Before this fix, the resolver was a `let` set once at init, so installAgent had no production write site to flip it. Every observer.record(...) call returned at the guard `guard let turnId = await turnIDResolver() else { return }`, and the four-seam AGENT-10 contract (SEC-07 dual-write + ME-04 channel + drain Task + ReplayLog writer) stayed dead for the process lifetime. Fix: - ReplayingToolResultObserver.turnIDResolver becomes a lock-protected slot (OSAllocatedUnfairLock — same pattern as InProcessConfirmationCache in JarvisMCP, async-safe in Swift 6 contexts unlike NSLock). attach(turnIDResolver:) replaces the slot; record(...) reads via withLock + invokes the resolver. - AgentOrchestrator gains a public actor-isolated `currentTurnID()` accessor over `currentTurn?.id`. One actor hop per tool dispatch is acceptable (tools fire 1–3× per turn). - AppDelegate.installAgent calls observer.attach(...) right after constructing the orchestrator, with `{ orchestrator?.currentTurnID() }` (weak capture, mirroring the broker/presenter lifetime invariant). A `systemLogger?.warning` fires if the observer isn't the expected concrete type (defense for future refactors). Tests added to MCPRuntimeWiringTests: - test_attach_turnIDResolver_flipsObserverFromDeadToLive — pre-attach record(...) drops; post-attach record(...) emits two envelopes correlated to the attached TurnID. - test_attach_replayLogReceivesSECDualWriteCorrelatedToTurnID — end-to-end through the production observer wiring: observer built with { nil }, attach(...) flips, record(...) drives the channel, drain Task writes to a real ephemeral ReplayLog. Reads the SQLite `events` table directly to assert exactly two `tool_result_full` rows landed for the post-attach toolUseId, both correlated to the attached TurnID, with zero rows for the pre-attach call. Refs audit-2026-05-12 CRIT-1 (.planning/audit-2026-05-12/mcp-agent-audit.md). Related: #21 (the dual-write was redundant in part because the dispatcher already 8-KB-capped what would have been the "full" raw side; that's fixed separately, but the SEC-07 audit trail is the non-redundant value of dual-write — pre-sanitize vs post-sanitize content for "did the model ever see content X?" forensics). Co-Authored-By: Claude Opus 4.7 --- App/AppDelegate.swift | 16 ++ App/MCP/ReplayingToolResultObserver.swift | 63 +++++-- .../AppTests/MCPRuntimeWiringTests.swift | 165 ++++++++++++++++++ .../AgentOrchestrator/AgentOrchestrator.swift | 10 ++ 4 files changed, 240 insertions(+), 14 deletions(-) diff --git a/App/AppDelegate.swift b/App/AppDelegate.swift index 471690c..5ef7299 100644 --- a/App/AppDelegate.swift +++ b/App/AppDelegate.swift @@ -1900,6 +1900,22 @@ final class AppDelegate: NSObject, NSApplicationDelegate { ) self.agentOrchestrator = orchestrator + // #20 (audit-2026-05-12 CRIT-1): wire the live `TurnID` resolver + // into `ReplayingToolResultObserver`. `MCPRuntimeWiring.build` + // constructed the observer with the `{ nil }` placeholder back in + // `applicationWillFinishLaunching` because the orchestrator didn't + // yet exist. Now that it does, flip the resolver to read + // `orchestrator.currentTurnID()`. Without this attach, the SEC-07 + // dual-write + ME-04 channel + drain Task stay dead in production + // (the observer's guard returns at every `record(...)` call). + if let observer = mcpRuntime.toolResultObserver as? ReplayingToolResultObserver { + observer.attach(turnIDResolver: { [weak orchestrator] in + await orchestrator?.currentTurnID() + }) + } else { + systemLogger?.warning("installAgent: toolResultObserver is not ReplayingToolResultObserver — turnIDResolver not wired; SEC-07 dual-write + ME-04 channel will stay dead") + } + // 4. Broadcaster + transcript store (D-05 / D-08 / BLOCKER-1). let broadcaster = OrchestratorEventBroadcaster(upstream: orchestrator.events) await broadcaster.start() diff --git a/App/MCP/ReplayingToolResultObserver.swift b/App/MCP/ReplayingToolResultObserver.swift index 5b88b91..32d2081 100644 --- a/App/MCP/ReplayingToolResultObserver.swift +++ b/App/MCP/ReplayingToolResultObserver.swift @@ -21,6 +21,7 @@ // in production traffic. import Foundation +import os.lock import AgentCore import Replay import JarvisMCP @@ -32,26 +33,58 @@ import Logging /// AppDelegate consumes from this channel and writes to ReplayLog. /// /// The observer needs a TurnID context to correlate replay rows with the -/// in-flight turn — the orchestrator owns that context, so the observer -/// holds a closure resolver that the orchestrator updates per-turn. In -/// the (current) AppDelegate boot — where AgentOrchestrator wiring lands -/// in Phase 6 voice / Phase 7 follow-ups — the resolver returns nil and -/// the observer logs without writing. The compile-time wiring is in -/// place; future plans flip the resolver to read the orchestrator's -/// active TurnID. +/// in-flight turn — the orchestrator owns that context. Since the observer +/// is constructed inside `MCPRuntimeWiring.build` (which runs BEFORE the +/// orchestrator is installed, because the orchestrator consumes the +/// dispatcher chain we return), the resolver is built with a placeholder +/// returning `nil`. After `installAgent` constructs the orchestrator, +/// AppDelegate calls `attach(turnIDResolver:)` to flip the resolver to a +/// closure that reads `orchestrator.currentTurnID()`. The lock-protected +/// holder mirrors `ConfirmationPresenterHolder.attach(_:)` — same pattern, +/// same lifecycle invariant. +/// +/// **#20 (audit-2026-05-12 CRIT-1):** before this fix, the resolver was a +/// `let` set once at init to `{ nil }`. The four-seam AGENT-10 contract +/// (SEC-07 dual-write + ME-04 channel + drain Task + ReplayLog writer) was +/// compiled but dead — every `record(...)` returned at the guard. Closing +/// the seam at install time is the production wiring fix. public final class ReplayingToolResultObserver: ToolResultObserver, @unchecked Sendable { + /// Box for the resolver closure so it can ride inside + /// `OSAllocatedUnfairLock`'s initialState (which requires `Sendable`, + /// and a bare closure type is fine; the box just shortens the type). + private struct ResolverBox: Sendable { + let resolver: @Sendable () async -> TurnID? + } + private let replayChannel: BoundedAsyncChannel - private let turnIDResolver: @Sendable () async -> TurnID? - private let logger: Logger + /// Lock-protected resolver slot. AppDelegate calls `attach(...)` + /// post-orchestrator-install to flip from the `{ nil }` placeholder to + /// the live closure. Uses `OSAllocatedUnfairLock` (the async-safe + /// scoped-locking primitive); same pattern as + /// `InProcessConfirmationCache` in JarvisMCP. + private let resolverSlot: OSAllocatedUnfairLock + private let logger: Logging.Logger public init( replayChannel: BoundedAsyncChannel, turnIDResolver: @escaping @Sendable () async -> TurnID? = { nil } ) { self.replayChannel = replayChannel - self.turnIDResolver = turnIDResolver - self.logger = Logger(label: JarvisLogChannel.replay.rawValue) + self.resolverSlot = OSAllocatedUnfairLock( + initialState: ResolverBox(resolver: turnIDResolver) + ) + self.logger = Logging.Logger(label: JarvisLogChannel.replay.rawValue) + } + + /// Replace the TurnID resolver. Called by AppDelegate once the + /// `AgentOrchestrator` is constructed inside `installAgent` so the + /// observer can correlate replay rows with the in-flight turn. + /// Mirrors `ConfirmationPresenterHolder.attach(_:)` — same lifecycle + /// invariant (observer outlives all calls; attach happens exactly once + /// per process). + public func attach(turnIDResolver: @escaping @Sendable () async -> TurnID?) { + resolverSlot.withLock { $0 = ResolverBox(resolver: turnIDResolver) } } public func record( @@ -61,9 +94,11 @@ public final class ReplayingToolResultObserver: ToolResultObserver, @unchecked S sanitizedBytes: Data ) async { // Resolve the active TurnID — production orchestrator wiring - // supplies this; pre-orchestrator boots return nil and the - // observer logs without writing (ReplayEvent requires a TurnID). - guard let turnId = await turnIDResolver() else { + // supplies this via `attach(turnIDResolver:)`; pre-attach calls + // return nil and the observer logs without writing (ReplayEvent + // requires a TurnID). + let resolver = resolverSlot.withLock { $0.resolver } + guard let turnId = await resolver() else { logger.debug("ToolResultObserver invoked without active TurnID — skipping replay write") return } diff --git a/App/Tests/AppTests/MCPRuntimeWiringTests.swift b/App/Tests/AppTests/MCPRuntimeWiringTests.swift index c497550..ed75875 100644 --- a/App/Tests/AppTests/MCPRuntimeWiringTests.swift +++ b/App/Tests/AppTests/MCPRuntimeWiringTests.swift @@ -309,6 +309,171 @@ final class MCPRuntimeWiringTests: XCTestCase { XCTAssertGreaterThanOrEqual(lastIdx, cap * 10 - cap - 5, "last drained envelope should be among the most recent bursts") } + + // MARK: - #20 (audit-2026-05-12 CRIT-1): turnIDResolver wiring + + /// #20 invariant 1: under the production construction shape — the + /// observer is built with the `{ nil }` placeholder resolver, then + /// `attach(turnIDResolver:)` flips it post-orchestrator-construction + /// — `record(...)` MUST produce envelopes into the channel after the + /// attach, and MUST stay silent before it. Pre-#20 the observer's + /// `turnIDResolver` was a `let` set once at init, so `attach(...)` + /// did not exist and production calls returned at the guard forever. + func test_attach_turnIDResolver_flipsObserverFromDeadToLive() async throws { + let channel = BoundedAsyncChannel(capacity: 64, policy: .dropOldest) + + // Mirror MCPRuntimeWiring.build's construction: placeholder { nil } + // resolver. This is what production set up before #20. + let observer = ReplayingToolResultObserver(replayChannel: channel) + + // Pre-attach record: the guard returns, nothing lands in the channel. + await observer.record( + toolUseId: "toolu_pre", + toolName: "get_time", + rawBytes: Data("RAW0".utf8), + sanitizedBytes: Data("SAN0".utf8) + ) + + // Drain non-blocking: peek the channel via a finish + iterate path. + // Since pre-attach should produce zero envelopes, finishing now and + // iterating must yield zero. + await channel.finish() + var preCount = 0 + for await _ in channel { preCount += 1 } + XCTAssertEqual(preCount, 0, + "pre-attach: observer with { nil } resolver must produce zero envelopes") + + // Now exercise the post-attach side with a fresh channel. + let liveChannel = BoundedAsyncChannel(capacity: 64, policy: .dropOldest) + let liveObserver = ReplayingToolResultObserver(replayChannel: liveChannel) + let expectedTurn = TurnID.fresh() + // Production wiring: AppDelegate's installAgent calls this with a + // closure that reads orchestrator.currentTurnID(). + liveObserver.attach(turnIDResolver: { expectedTurn }) + + await liveObserver.record( + toolUseId: "toolu_post", + toolName: "get_time", + rawBytes: Data("RAW".utf8), + sanitizedBytes: Data("SAN".utf8) + ) + + var liveIter = liveChannel.makeAsyncIterator() + let first = await liveIter.next() + let second = await liveIter.next() + guard + case .toolResultFull(let id1, let bytes1) = first?.event, + case .toolResultFull(let id2, let bytes2) = second?.event + else { + XCTFail("post-attach: expected two .toolResultFull envelopes") + return + } + XCTAssertEqual(id1, "toolu_post") + XCTAssertEqual(bytes1, Data("SAN".utf8)) + XCTAssertEqual(id2, "toolu_post:raw") + XCTAssertEqual(bytes2, Data("RAW".utf8)) + XCTAssertEqual(first?.turnId, expectedTurn, + "envelope must carry the attached resolver's TurnID") + XCTAssertEqual(second?.turnId, expectedTurn) + } + + /// #20 invariant 2: end-to-end through the production observer wiring — + /// observer constructed with the `{ nil }` placeholder (like + /// `MCPRuntimeWiring.build` does), then `attach(turnIDResolver:)` flips + /// it (like `installAgent` does), then a simulated dispatcher call + /// `record(...)` drives the SEC-07 dual-write into the channel; the + /// drain Task pulls envelopes onto a real ephemeral `ReplayLog` and we + /// assert by reading the SQLite `events` table directly that both the + /// sanitized and the `:raw` rows landed correlated to the attached + /// `TurnID`. Pre-#20 the guard returned, the channel stayed empty, and + /// the table had zero `tool_result_full` rows. + func test_attach_replayLogReceivesSECDualWriteCorrelatedToTurnID() async throws { + let tmpDB = FileManager.default.temporaryDirectory + .appendingPathComponent("issue20-\(UUID().uuidString).sqlite") + defer { try? FileManager.default.removeItem(at: tmpDB) } + + let log = try ReplayLog(databaseURL: tmpDB) + let sessionId = try await log.beginSession(appVersion: "test", buildSHA: "abc") + let turnId = TurnID.fresh() + try await log.startTurn( + turnId: turnId, sessionId: sessionId, retryOf: nil, + turnNonce: UUID().uuidString, source: .text, + provider: "test", modelId: "test" + ) + + // Production-shape composition. The observer is built with the + // `{ nil }` placeholder, matching `MCPRuntimeWiring.build`'s + // construction-time call. Pre-attach, the observer's guard returns + // at every `record(...)` call. + let channel = BoundedAsyncChannel(capacity: 64, policy: .dropOldest) + let observer = ReplayingToolResultObserver(replayChannel: channel) + + // Mirror the AppDelegate drain Task — pulls envelopes off the + // channel and writes them to ReplayLog under the per-envelope TurnID. + let drainTask = Task.detached { + for await env in channel { + await log.record(env.event, for: env.turnId) + } + } + + // STEP 1: pre-attach record. Simulates a dispatcher's + // `observer.record(...)` call before `installAgent` had a chance + // to wire the resolver. The observer logs and returns; nothing + // lands in the channel. + await observer.record( + toolUseId: "pre-id", + toolName: "get_time", + rawBytes: Data("raw-pre".utf8), + sanitizedBytes: Data("san-pre".utf8) + ) + + // STEP 2: simulate installAgent's wiring — attach a resolver that + // yields the known `turnId` (the production closure reads + // `orchestrator.currentTurnID()` instead). + observer.attach(turnIDResolver: { turnId }) + + // STEP 3: post-attach record. The same code path; this time the + // resolver yields a non-nil TurnID and envelopes flow through the + // channel to the drain Task to ReplayLog. + await observer.record( + toolUseId: "live-id", + toolName: "get_time", + rawBytes: Data("raw-live".utf8), + sanitizedBytes: Data("san-live".utf8) + ) + + // Allow drain + ReplayLog batch flush. + try await Task.sleep(nanoseconds: 300_000_000) + await channel.finish() + _ = await drainTask.value + await log.endTurn(turnId, stopReason: "test") + await log.flush() + try await log.close() + + // Open the SQLite file directly and assert exactly the post-attach + // envelopes landed (two: sanitized + `:raw`). Pre-attach must have + // produced zero rows for `pre-id` because the guard returned. + let conn = try SQLiteConnection.open(at: tmpDB) + defer { try? conn.close() } + let rows: [(turnId: String, payload: Data)] = try conn.query( + "SELECT turn_id, payload_bytes FROM events WHERE kind='tool_result_full';", + map: { ($0.columnText(at: 0) ?? "", $0.columnBlob(at: 1) ?? Data()) } + ) + XCTAssertEqual(rows.count, 2, + "post-attach: SEC-07 dual-write produces 2 rows (sanitized + :raw); pre-attach produced 0") + for r in rows { + XCTAssertEqual(r.turnId, turnId.rawValue, + "every row must carry the attached resolver's TurnID") + } + let ids: [String] = rows.compactMap { row in + guard let json = try? JSONSerialization.jsonObject(with: row.payload) as? [String: String] else { + return nil + } + return json["tool_use_id"] + }.sorted() + XCTAssertEqual(ids, ["live-id", "live-id:raw"], + "rows must reference the post-attach record's toolUseId — pre-attach `pre-id` was silently dropped") + } } // Actor extension to set broker — pattern matches the JarvisMCP test fixtures. diff --git a/packages/AgentCore/Sources/AgentOrchestrator/AgentOrchestrator.swift b/packages/AgentCore/Sources/AgentOrchestrator/AgentOrchestrator.swift index f813524..470894e 100644 --- a/packages/AgentCore/Sources/AgentOrchestrator/AgentOrchestrator.swift +++ b/packages/AgentCore/Sources/AgentOrchestrator/AgentOrchestrator.swift @@ -206,6 +206,16 @@ public actor AgentOrchestrator { voiceOriginatedTurns.contains(turnId) } + /// #20 (audit-2026-05-12 CRIT-1) — the in-flight `TurnID`, or `nil` if + /// no turn is active. `ReplayingToolResultObserver.attach(...)` reads + /// this from a `@Sendable () async -> TurnID?` closure so the SEC-07 + /// dual-write + ME-04 channel can correlate replay rows with the + /// current turn. One actor-hop per tool dispatch — acceptable cost + /// (tools fire 1–3× per turn). + public func currentTurnID() -> TurnID? { + currentTurn?.id + } + /// BLOCKER-2 actor-internal populator. Called from `runTurn` after /// allocating the `TurnID` when `input.source == .voice`. private func recordVoiceTurn(_ turnId: TurnID) { -- 2.54.0