diff --git a/docs/reference/terminal-sessions.md b/docs/reference/terminal-sessions.md index a25b09355..81fafb819 100644 --- a/docs/reference/terminal-sessions.md +++ b/docs/reference/terminal-sessions.md @@ -417,3 +417,46 @@ retain `follow_view_required` in a 200 `ok:false` body, as before. A changed preview remains `preview_changed` (409). During automatic refresh a missing or ambiguous scope retains the existing tiles and names the source problem in the explanation strip; refresh or reopen it under the current unambiguous key. + + +## Reading or sending to an expected native conversation + +An application that selects an original Claude Code or Codex conversation can opt +into an expected-identity check. The identity is the exact tuple `nodeId`, canonical +`sessionName`, `runtime` (`claude-code` or `codex`) and native `nativeId`. A registry +resume token by itself is not a current-process identity observation. + +Read `GET /api/sessions/:sessionName/generation-record-expected` with query fields +`expectedNodeId`, `expectedRuntime`, `expectedNativeId` and optional `sinceBytes`. +It resolves the verified current occupant and native process, opens that native +record, checks its identity, and returns identity and bounded suffix bytes from +the same opened file. Its existing byte cursor and `generationId` semantics match +`generation-record`. Success adds `identityContract:1` and `checkedIdentity`, the +exact expected tuple. Check both before displaying the suffix as that conversation. + +Send `POST /api/transport/send-expected` with the ordinary send body and a required +`expectedIdentity` object containing those four fields. The check runs again after +buffer preparation immediately before paste, and before Enter after any wait. +A retained input stores the constraint in its outbox receipt; committed dispatch +and reuse of its delivery ID cannot select a different original conversation or +omit the stored constraint. Delivery IDs still do not deduplicate explicit input. +Success adds `identityContract:1` and `checkedIdentity`. Retained is not sent; +`checkedIdentity` is an observation at an input boundary, not model consumption. +If replacement is observed after paste, the result reports `inputPhase:"written"` +and `sent:true`, with Enter refused. Inspect that input before deciding what to do. + +Malformed expectations return 400 (`invalid_expected_identity`). A different +observed identity returns 409 (`expected_identity_mismatch`); a missing or +unverified identity returns 409 (`expected_identity_unavailable`). Conflicting +reuse of a delivery ID returns 409 (`delivery_identity_conflict`). No conditional +operation falls back to the replacement. Use these dedicated paths when detecting +support: older daemons return 404 instead of silently ignoring an unknown field. +The ordinary read and send endpoints keep their existing behavior for callers +that do not opt in. + +This is a finite process/record observation followed by a terminal write; tmux has +no atomic native-conversation compare-and-write operation. A native conversation +change after the final observation remains a race. The check does not turn a +terminal attachment into a conversation pin: existing terminal viewers follow +managed pane/session bindings and may display their replacement. Applications +must keep that attachment limit separate from checked record reads and sends. diff --git a/packages/daemon/src/adapters/tmux.ts b/packages/daemon/src/adapters/tmux.ts index f70e27a26..8e2f9c8db 100644 --- a/packages/daemon/src/adapters/tmux.ts +++ b/packages/daemon/src/adapters/tmux.ts @@ -1021,11 +1021,12 @@ export class TmuxAdapter { } } - async sendKeys(target: string, keys: string[], beforeInput?: () => void): Promise { + async sendKeys(target: string, keys: string[], beforeInput?: () => void, options?: { beforeKeys?: () => Promise }): Promise { return this.guardedInput(target, async (pane, beforeWrite) => { const identity = this.freshManaged.get(target)?.nodeId ?? target; const submit = keys.some(key => key === "Enter" || key === "C-m"); if (this.deliveryGuard?.needsInputCheck(identity)) await this.deliveryGuard.beforeInput(identity, submit); + if (options?.beforeKeys) await options.beforeKeys(); beforeWrite(); beforeInput?.(); const result = await this.sendKeysUnchecked(pane, keys); if (result.ok && submit) this.deliveryGuard?.noteSubmitted(identity); diff --git a/packages/daemon/src/domain/draft-aware-delivery.ts b/packages/daemon/src/domain/draft-aware-delivery.ts index 9cb536a1e..a7acadb29 100644 --- a/packages/daemon/src/domain/draft-aware-delivery.ts +++ b/packages/daemon/src/domain/draft-aware-delivery.ts @@ -10,7 +10,7 @@ export interface GuardedDeliverySummary { } /** Historical attempts contain outcomes, never executable options or callbacks. */ -interface Receipt { ids: string[]; result?: SendResult } +interface Receipt { ids: string[]; result?: SendResult; expectedIdentity?: import("./expected-conversation.js").ExpectedConversation } type Deliver = (session: string, text: string, opts: SendOpts) => Promise; /** Applicability needs the intended queue row and recipient, not a saved delivery receipt. */ export type QueueWakeFacts = Pick; @@ -42,7 +42,7 @@ export class DraftAwareDelivery { /** The existing per-seat lease serializes requests. Each current request gets an attempt, not an old result. */ async send(session: string, text: string, opts: SendOpts, ids: string[]): Promise { if (!ids.length || new Set(ids).size !== ids.length) throw new DeliveryGuardError("invalid_delivery_ids", "Distinct delivery IDs are required."); - const pending = this.attempt(session, text, opts, { ids }); + const pending = this.attempt(session, text, opts, { ids, ...(opts.expectedIdentity ? { expectedIdentity: opts.expectedIdentity } : {}) }); this.sends.add(pending); try { return await pending; } finally { this.sends.delete(pending); } @@ -69,7 +69,10 @@ export class DraftAwareDelivery { } })(); recorded = true; - } catch { /* Bookkeeping cannot veto input; report its absence with the actual result. */ } + } catch (error) { + if (opts.expectedIdentity) throw error; // A conditional send cannot lose its durable constraint and still type. + /* Legacy bookkeeping cannot veto input; report its absence with the actual result. */ + } try { await this.guard.beforeInput(session); result = await this.deliver(session, text, { ...opts, diff --git a/packages/daemon/src/domain/expected-conversation.ts b/packages/daemon/src/domain/expected-conversation.ts new file mode 100644 index 000000000..fe2aecce1 --- /dev/null +++ b/packages/daemon/src/domain/expected-conversation.ts @@ -0,0 +1,88 @@ +import { statSync } from "node:fs"; +import type { SessionRegistry } from "./session-registry.js"; +import type { TmuxAdapter } from "../adapters/tmux.js"; +import type { ContextUsageStore } from "./context-usage-store.js"; +import { SeatIdentityStore } from "./seat-identity-store.js"; +import { ProcessCensus } from "./process-census.js"; +import { CodexThreadIdResolver } from "./codex-thread-id.js"; +import { resolveIdentityVerifiedClaudeRecord, resolveLiveCodexThreadId } from "./model-divergence/current-generation-record.js"; +import { parseSqliteUtcMs } from "./sqlite-time.js"; +import { DeliveryGuardError } from "./seat-delivery-guard.js"; + +export interface ExpectedConversation { + nodeId: string; + sessionName: string; + runtime: "claude-code" | "codex"; + nativeId: string; +} +const census = new ProcessCensus({ freshnessMs: 0 }); +const threads = new CodexThreadIdResolver({ codexHome: process.env.CODEX_HOME || undefined }); +function unavailable(message: string): never { throw new DeliveryGuardError("expected_identity_unavailable", message); } +function mismatch(message: string): never { throw new DeliveryGuardError("expected_identity_mismatch", message); } + +export function expectedConversation(value: unknown): ExpectedConversation { + if (!value || typeof value !== "object" || Array.isArray(value)) throw new DeliveryGuardError("invalid_expected_identity", "An expected conversation identity is required."); + const raw = value as Record; + for (const key of ["nodeId", "sessionName", "nativeId"]) { + if (typeof raw[key] !== "string" || !(raw[key] as string).length || (raw[key] as string).length > 256 || /\s/.test(raw[key] as string)) + throw new DeliveryGuardError("invalid_expected_identity", `Expected identity needs a nonempty ${key}, without whitespace.`); + } + if (raw.runtime !== "claude-code" && raw.runtime !== "codex") throw new DeliveryGuardError("invalid_expected_identity", "Expected identity supports claude-code or codex only."); + return { nodeId: raw.nodeId as string, sessionName: raw.sessionName as string, runtime: raw.runtime, nativeId: raw.nativeId as string }; +} +export function sameConversation(a: ExpectedConversation, b: ExpectedConversation): boolean { + return a.nodeId === b.nodeId && a.sessionName === b.sessionName && a.runtime === b.runtime && a.nativeId === b.nativeId; +} + +/** Opt-in consumer check; a registry resume token alone never authorizes the operation. */ +export async function resolveExpectedConversation(expected: ExpectedConversation, registry: SessionRegistry, tmux: TmuxAdapter, + store: ContextUsageStore | undefined): Promise<{ identity: ExpectedConversation; path: string; generation: string; pane: string; assertCurrent: () => void }> { + if (!store) unavailable("No native record resolver is available; no conditional input or record is served."); + const context = registry.findResumeContextByName(expected.sessionName); + if (!context) unavailable("The expected seat has no registered conversation."); + if (context.nodeId !== expected.nodeId || context.runtime !== expected.runtime) mismatch("The seat or runtime is no longer the expected one."); + const occupant = registry.currentOccupantTenure(context.nodeId), binding = registry.getBindingForNode(context.nodeId); + const proof = new SeatIdentityStore(registry.db).getForNode(context.nodeId); + if (!occupant || !binding?.tmuxPane || binding.tmuxSession !== expected.sessionName || proof?.verdict !== "verified" + || proof.sessionName !== expected.sessionName || proof.evidence.registeredPane !== binding.tmuxPane + || !(Date.parse(proof.observedAt) >= parseSqliteUtcMs(occupant.bootAt))) unavailable("The current occupant/pane identity is unverified."); + const pane = binding.tmuxPane, generation = occupant.generationUuid; + const assertCurrent = () => { + const current = registry.findResumeContextByName(expected.sessionName), bound = registry.getBindingForNode(expected.nodeId); + if (current?.nodeId !== expected.nodeId || current.runtime !== expected.runtime || bound?.tmuxSession !== expected.sessionName + || bound.tmuxPane !== pane || registry.currentOccupantTenure(expected.nodeId)?.generationUuid !== generation) + mismatch("The occupant binding changed during the conditional operation."); + }; + let nativeId: string, path: string | null, rootPid: number | null; + try { + rootPid = await tmux.getPanePid(pane); + if (!rootPid || rootPid !== (proof.evidence.observedPanePid ?? proof.evidence.observedPid)) unavailable("The pane process no longer matches its verified occupant."); + const deps = { getPanePid: () => tmux.getPanePid(pane), listProcesses: () => census.list(), + readThreadIdByPid: async (pid: number, startedAt?: string) => startedAt ? threads.resolve(pid, startedAt) : undefined }; + if (expected.runtime === "claude-code") { + const sidecar = store.readSidecar(expected.sessionName); + const selected = await resolveIdentityVerifiedClaudeRecord({ sessionName: expected.sessionName, generation, + occupantBootAt: occupant.bootAt, binding, identity: proof, sidecar: sidecar.ok ? sidecar.data : null }, deps, + file => { try { return statSync(file).isFile(); } catch { return false; } }); + if (!selected.ok) unavailable(selected.reason); + nativeId = selected.id; path = selected.path; + } else { + let verifiedThread: string | undefined; + const selected = await resolveLiveCodexThreadId(pane, { ...deps, readThreadIdByPid: async (pid, startedAt) => { + const id = await deps.readThreadIdByPid(pid, startedAt); if (pid === proof.evidence.observedPid) verifiedThread = id; return id; + } }); + if (!selected.ok) unavailable(selected.reason); + nativeId = selected.id; path = store.readCodexTranscriptPath(nativeId); + if (proof.evidence.observedPanePid != null && verifiedThread !== nativeId) unavailable("The selected thread is not owned by the verified native process."); + } + assertCurrent(); + if (await tmux.getPanePid(pane) !== rootPid) mismatch("The pane process changed during native resolution."); + assertCurrent(); + } catch (error) { + if (error instanceof DeliveryGuardError) throw error; + unavailable(`Native identity could not be resolved: ${error instanceof Error ? error.message : String(error)}`); + } + if (nativeId! !== expected.nativeId) mismatch("The current native conversation differs from the expected original."); + if (!path!) unavailable("The expected native record is unavailable."); + return { identity: { ...expected }, path: path!, generation, pane, assertCurrent }; +} diff --git a/packages/daemon/src/domain/outbox-handler.ts b/packages/daemon/src/domain/outbox-handler.ts index c0eb02291..964a30293 100644 --- a/packages/daemon/src/domain/outbox-handler.ts +++ b/packages/daemon/src/domain/outbox-handler.ts @@ -1,3 +1,4 @@ +import type { ExpectedConversation } from "./expected-conversation.js"; import type Database from "better-sqlite3"; /** @@ -114,12 +115,36 @@ export class OutboxHandler { * era-stamp column, so the writer degrades (omits it) instead of throwing. */ private readonly hasIdentityProvenanceColumn: boolean; private readonly hasIdentityPrincipalColumn: boolean; + private readonly hasGuardDeliveryColumn: boolean; constructor(db: Database.Database) { this.db = db; const columns = (this.db.prepare("PRAGMA table_info(outbox_entries)").all() as Array<{ name: string }>).map((col) => col.name); this.hasIdentityProvenanceColumn = columns.includes("identity_provenance"); this.hasIdentityPrincipalColumn = columns.includes("identity_principal"); + this.hasGuardDeliveryColumn = columns.includes("guard_delivery"); + } + + /** The optional original-conversation constraint shares the existing durable delivery receipt. */ + expectedIdentity(outboxId: string): ExpectedConversation | undefined { + if (!this.hasGuardDeliveryColumn) return undefined; + const row = this.db.prepare("SELECT guard_delivery FROM outbox_entries WHERE outbox_id=?").get(outboxId) as { guard_delivery: string | null } | undefined; + return row?.guard_delivery ? JSON.parse(row.guard_delivery).expectedIdentity : undefined; + } + + bindExpectedIdentity(input: OutboxRecordInput & { outboxId: string }, expected: ExpectedConversation): void { + if (!this.hasGuardDeliveryColumn) throw new OutboxHandlerError("expected_identity_unavailable", "The durable conditional-delivery schema is unavailable."); + this.db.transaction(() => { + const prior = this.getById(input.outboxId), bound = this.expectedIdentity(input.outboxId); + if (prior && (prior.senderSession !== input.senderSession || prior.destinationSession !== input.destinationSession || prior.body !== input.body) + || bound && ["nodeId", "sessionName", "runtime", "nativeId"].some(key => bound[key as keyof ExpectedConversation] !== expected[key as keyof ExpectedConversation])) + throw new OutboxHandlerError("delivery_identity_conflict", "This delivery ID already selects different input or an original conversation."); + if (!prior) this.record(input); + const row = this.db.prepare("SELECT guard_delivery FROM outbox_entries WHERE outbox_id=?").get(input.outboxId) as { guard_delivery: string | null }; + const receipt = row.guard_delivery ? JSON.parse(row.guard_delivery) : { ids: [input.outboxId] }; + this.db.prepare("UPDATE outbox_entries SET guard_delivery=? WHERE outbox_id=?") + .run(JSON.stringify({ ...receipt, expectedIdentity: expected }), input.outboxId); + })(); } record(input: OutboxRecordInput): OutboxEntry { diff --git a/packages/daemon/src/domain/session-transport.ts b/packages/daemon/src/domain/session-transport.ts index 814732432..e8740467b 100644 --- a/packages/daemon/src/domain/session-transport.ts +++ b/packages/daemon/src/domain/session-transport.ts @@ -1,3 +1,5 @@ +import { expectedConversation, sameConversation, resolveExpectedConversation, type ExpectedConversation } from "./expected-conversation.js"; +import type { ContextUsageStore } from "./context-usage-store.js"; import { physicalSessionName } from "./session-registry.js"; import { currentClaudeMcpMenu } from "./claude-mcp-menu.js"; import { currentClaudeResumePicker } from "./native-resume-probe.js"; @@ -1718,6 +1720,7 @@ export function readInputBox(pane: string | null, runtime: "claude-code" | "code } export interface SendOpts { + expectedIdentity?: ExpectedConversation; /** Internal provenance derived by the transport route. */ identityProvenance?: string; /** Internal: with identityProvenance principal:v1, the MCP adapter's caller (the route's resolveRecordedIdentity). */ @@ -1857,6 +1860,8 @@ export interface ExistingInput { type OverrideInputAudit = { classification: ExistingInput["classification"]; visibleRows: number; suggestionRows: number; choice: "inspect" | "type_without_checking"; outcome: "stopped" | "proceed"; clearAttempted: false }; export interface SendResult { + checkedIdentity?: ExpectedConversation; + identityContract?: 1; existingInput?: ExistingInput; ok: boolean; sessionName: string; @@ -1959,6 +1964,7 @@ export interface TransportReadinessChecker { } interface SessionTransportDeps { + contextUsageStore?: ContextUsageStore; db: Database.Database; rigRepo: RigRepository; sessionRegistry: SessionRegistry; @@ -2014,6 +2020,7 @@ interface SessionReconciliationSnapshot { } export class SessionTransport { + private contextUsageStore?: ContextUsageStore; readonly db: Database.Database; readonly guardedDelivery?: DraftAwareDelivery; private rigRepo: RigRepository; @@ -2034,6 +2041,7 @@ export class SessionTransport { private verifyInputBox?: () => boolean; constructor(deps: SessionTransportDeps) { + this.contextUsageStore = deps.contextUsageStore; this.db = deps.db; this.rigRepo = deps.rigRepo; this.sessionRegistry = deps.sessionRegistry; @@ -2426,6 +2434,57 @@ export class SessionTransport { } private async sendClassified(sessionName: string, text: string, opts?: SendOpts): Promise { + // The constraint belongs to the durable input, including later committed dispatch and ID reuse. + const outbox = new OutboxHandler(this.db); + let ids = opts?.committedOutboxIds ?? (opts?.deliveryId ? [opts.deliveryId] : []); + try { + let expected = opts?.expectedIdentity ? expectedConversation(opts.expectedIdentity) : undefined; + for (const id of ids) { + const saved = outbox.expectedIdentity(id); + if (!saved) continue; + const bound = expectedConversation(saved); + if (expected && !sameConversation(expected, bound)) throw new DeliveryGuardError("delivery_identity_conflict", "This input ID is bound to another original conversation."); + expected = bound; + } + if (!expected) return this.sendClassifiedInput(sessionName, text, opts); + if (sessionName !== expected.sessionName) throw new DeliveryGuardError("expected_identity_mismatch", "The destination must be the expected canonical session."); + if (!ids.length) ids = [`expected-send-${randomUUID()}`]; + this.db.transaction(() => { + for (const id of ids) { + const prior = opts?.committedOutboxIds ? outbox.getById(id) : null; + if (opts?.committedOutboxIds && !prior) throw new DeliveryGuardError("delivery_identity_conflict", "Committed input ID is missing."); + outbox.bindExpectedIdentity(prior ? { ...prior, outboxId: id, tags: prior.tags ?? undefined, auditPointer: prior.auditPointer ?? undefined } : { + outboxId: id, senderSession: opts?.actorSession ?? "unknown", destinationSession: sessionName, body: text, + identityProvenance: opts?.identityProvenance, identityPrincipal: opts?.identityPrincipal, + }, expected!); + } + })(); + await resolveExpectedConversation(expected, this.sessionRegistry, this.tmuxAdapter, this.contextUsageStore); + const result = await this.sendClassifiedInput(sessionName, text, { ...opts, expectedIdentity: expected, + ...(opts?.committedOutboxIds ? {} : { deliveryId: ids[0] }) }); + if (result.outcome !== "retained") { + try { + for (const id of ids) { + if (result.ok) outbox.markDelivered(id); + else if (result.sent === false || result.inputPhase === "not_reached") outbox.markFailed(id); + else outbox.markIndeterminate(id); + } + } catch { result.warning = [result.warning, "Conditional input receipt could not be finalized; inspect before retrying."].filter(Boolean).join(" "); } + } + return { ...result, identityContract: 1, outboxIds: result.outboxIds ?? ids }; + } catch (error) { + // Before-input conditional failures never fall back to an ordinary send. + const written = (error as { inputWritten?: boolean }).inputWritten; + const attempted = (error as { inputAttempted?: boolean }).inputAttempted; + if (!written && !attempted) for (const id of ids) { try { if (outbox.getById(id)) outbox.markFailed(id); } catch { /* Original refusal remains authoritative. */ } } + return { ok: false, sessionName, identityContract: 1, outboxIds: ids, + sent: written ? true : attempted ? undefined : false, + inputPhase: written ? "written" : attempted ? "attempted" : "not_reached", + reason: (error as { code?: string }).code ?? "expected_identity_unavailable", error: (error as Error).message }; + } + } + + private async sendClassifiedInput(sessionName: string, text: string, opts?: SendOpts): Promise { const guard = this.tmuxAdapter.deliveryGuard; const admission: AdmissionInput = { source: opts?.openrigNotice ? "openrig" : "agent", urgent: !!opts?.urgent, override: opts?.dangerouslyInteract === true }; if (!guard) { @@ -2497,7 +2556,7 @@ export class SessionTransport { if (this.guardedDelivery && !memberOpts?.submitOnly && !memberOpts?.dangerouslyInteract && guard.needsInputCheck(sessionName)) { if (callerSignal(memberOpts)?.aborted) return callerCancelled(sessionName); return this.guardedDelivery.send(sessionName, memberText, memberOpts ?? {}, - memberOpts?.committedOutboxIds ?? [`guard-send-${randomUUID()}`]); + memberOpts?.committedOutboxIds ?? [memberOpts?.deliveryId ?? `guard-send-${randomUUID()}`]); } return this.sendUnguarded(sessionName, memberText, memberOpts); }, async target => { @@ -2701,6 +2760,13 @@ export class SessionTransport { sessionName = physicalSessionName(this.db, sessionName); let preVerifyContent: string | null = null; const sessionMeta = this.getSessionMeta(sessionName); + let checkedIdentity: ExpectedConversation | undefined; + let originalCheck: Awaited> | undefined; + const checkOriginal = opts?.expectedIdentity ? async () => { + originalCheck = await resolveExpectedConversation(opts.expectedIdentity!, this.sessionRegistry, this.tmuxAdapter, this.contextUsageStore); + checkedIdentity = originalCheck.identity; + } : undefined; + const beforeOriginalInput = () => { originalCheck?.assertCurrent(); opts?.beforeWrite?.(); }; const runtime = sessionMeta.runtime; const readTarget = sessionMeta.pane ?? sessionName; let runtimeAdvisory: string | undefined; @@ -2749,6 +2815,7 @@ export class SessionTransport { let existingInput: ExistingInput | undefined; let overrideBeforePaste: (() => Promise) | undefined; const observe = (result: SendResult): SendResult => { + if (checkedIdentity) result = { ...result, checkedIdentity, identityContract: 1 }; if (existingInput) result = { ...result, existingInput }; if (result.ok && runtimeAdvisory) result = { ...result, warning: [runtimeAdvisory, result.warning].filter(Boolean).join(" ") }; if (observed && this.captureObserver) { @@ -2903,7 +2970,8 @@ export class SessionTransport { try { submitResult = await this.runStage( "session_transport.submit", - () => beforeKey ? this.tmuxAdapter.sendKeys(sessionName, ["Enter"], beforeKey) : this.tmuxAdapter.sendKeys(sessionName, ["Enter"]), + () => checkOriginal ? this.tmuxAdapter.sendKeys(sessionName, ["Enter"], () => { beforeOriginalInput(); beforeKey?.(); }, { beforeKeys: checkOriginal }) + : beforeKey ? this.tmuxAdapter.sendKeys(sessionName, ["Enter"], beforeKey) : this.tmuxAdapter.sendKeys(sessionName, ["Enter"]), (result) => result.ok ? "ok" : "failed", ); } catch (error) { @@ -2913,7 +2981,7 @@ export class SessionTransport { } if (!submitResult.ok && submitResult.code === CALLER_CANCELLED && cancel?.aborted) return callerCancelled(sessionName); if (!submitResult.ok) { - return { ok: false, sessionName, reason: "submit_failed", outcome: "failed", error: `submitOnly: Enter did not land on '${sessionName}': ${submitResult.message}` }; + return { ok: false, sessionName, ...(submitResult.code?.startsWith("expected_identity_") ? { sent: false, inputPhase: "not_reached" as const } : {}), reason: submitResult.code?.startsWith("expected_identity_") ? submitResult.code : "submit_failed", outcome: "failed", error: `submitOnly: Enter did not land on '${sessionName}': ${submitResult.message}` }; } try { opts.onInputEffect?.("submitted"); } catch { /* Observation cannot change an input result. */ } return observe({ ok: true, sessionName, outcome: "rendered-unconfirmed", submitOnly: true }); @@ -3217,6 +3285,9 @@ export class SessionTransport { () => { opts?.beforeWrite?.(); writeIssued = true; + if (checkOriginal) return this.tmuxAdapter.sendText(sessionName, text, () => { beforeOriginalInput(); beforePaste?.(); }, { + ...(promptOverride ? { bracketed: false } : {}), beforePaste: async () => { await overrideBeforePaste?.(); await checkOriginal(); }, + }); if (promptOverride) return this.tmuxAdapter.sendText(sessionName, text, beforePaste, { bracketed: false, beforePaste: overrideBeforePaste }); return beforePaste ? this.tmuxAdapter.sendText(sessionName, text, beforePaste) : this.tmuxAdapter.sendText(sessionName, text); }, @@ -3227,7 +3298,7 @@ export class SessionTransport { } if (!textResult.ok) { // A draft refusal is the guard holding the write before any input; it is not an attempted write. - const policyRefusal = typeof textResult.code === "string" && (textResult.code.startsWith("draft_") || textResult.code.startsWith("override_input_") || textResult.code === "managed_compaction_pending"); + const policyRefusal = typeof textResult.code === "string" && (textResult.code.startsWith("expected_identity_") || textResult.code.startsWith("draft_") || textResult.code.startsWith("override_input_") || textResult.code === "managed_compaction_pending"); return observe({ ok: false, sessionName, @@ -3279,11 +3350,15 @@ export class SessionTransport { // 5. Submit (Enter) submitResult = await this.runStage( "session_transport.submit", - () => { opts?.beforeWrite?.(); enterIssued = true; return opts?.beforeWrite ? this.tmuxAdapter.sendKeys(sessionName, ["Enter"], opts.beforeWrite) : this.tmuxAdapter.sendKeys(sessionName, ["Enter"]); }, + () => { + if (checkOriginal) return this.tmuxAdapter.sendKeys(sessionName, ["Enter"], () => { beforeOriginalInput(); enterIssued = true; }, { beforeKeys: checkOriginal }); + opts?.beforeWrite?.(); enterIssued = true; + return opts?.beforeWrite ? this.tmuxAdapter.sendKeys(sessionName, ["Enter"], opts.beforeWrite) : this.tmuxAdapter.sendKeys(sessionName, ["Enter"]); + }, (result) => result.ok ? "ok" : "failed", ); if (!submitResult.ok) { - const policyRefusal = typeof submitResult.code === "string" && (submitResult.code.startsWith("draft_") || submitResult.code === "managed_compaction_pending"); + const policyRefusal = typeof submitResult.code === "string" && (submitResult.code.startsWith("expected_identity_") || submitResult.code.startsWith("draft_") || submitResult.code === "managed_compaction_pending"); return observe({ ok: false, sessionName, diff --git a/packages/daemon/src/routes/sessions.ts b/packages/daemon/src/routes/sessions.ts index 71886ebe1..ad01cf4bc 100644 --- a/packages/daemon/src/routes/sessions.ts +++ b/packages/daemon/src/routes/sessions.ts @@ -1,3 +1,5 @@ +import type { Context } from "hono"; +import { expectedConversation, resolveExpectedConversation, type ExpectedConversation } from "../domain/expected-conversation.js"; import { unknownModelObservation, type CurrentModelReader } from "../domain/model-divergence/current-model-reader.js"; import { inventoryCaptureOptions, type ShadowCapture } from "../domain/shadow-capture.js"; import { Hono } from "hono"; @@ -731,11 +733,17 @@ sessionAdminRoutes.post("/:sessionRef/unclaim", terminalAuthGuard(), async (c) = // impossible, never receive an empty success. // Round-2 (r2 HIGH-2): raw conversation bytes with no transcript redaction — a TERMINAL-CLASS // surface behind the same bearer gate as its neighbors (401/401/200; null-token loopback passes). -sessionAdminRoutes.get("/:sessionName/generation-record", terminalAuthGuard(), async (c) => { +const generationRecord = async (c: Context) => { const sessionName = c.req.param("sessionName")!; + let expected: ExpectedConversation | undefined; + if (c.req.path.endsWith("/generation-record-expected")) { + try { expected = expectedConversation({ nodeId: c.req.query("expectedNodeId"), sessionName, + runtime: c.req.query("expectedRuntime"), nativeId: c.req.query("expectedNativeId") }); } + catch (error) { return c.json({ error: "invalid_expected_identity", message: (error as Error).message }, 400); } + } const store = c.get("contextUsageStore" as never) as ContextUsageStore | undefined; if (!store) { - return c.json({ error: "unsupported_runtime", message: "No context-usage store on this daemon; the seat's generation record cannot be resolved." }, 409); + return c.json({ error: expected ? "expected_identity_unavailable" : "unsupported_runtime", message: "No context-usage store on this daemon; the seat's generation record cannot be resolved." }, 409); } const { sessionRegistry: registry, tmuxAdapter } = getDeps(c); const context = registry?.findResumeContextByName(sessionName); @@ -743,7 +751,12 @@ sessionAdminRoutes.get("/:sessionName/generation-record", terminalAuthGuard(), a const runtime = context?.runtime ?? "claude-code"; let transcriptPath: string | null; let sessionId: string | null; - if (runtime === "codex") { + let original: Awaited> | undefined; + if (expected) { + try { original = await resolveExpectedConversation(expected, registry, tmuxAdapter, store); + sessionId = original.identity.nativeId; transcriptPath = original.path; + } catch (error) { return c.json({ error: (error as { code?: string }).code ?? "expected_identity_unavailable", message: (error as Error).message }, 409); } + } else if (runtime === "codex") { const binding = registry.getBindingForNode(context!.nodeId); const identity = new SeatIdentityStore(registry.db).getForNode(context!.nodeId); if (!occupant || !binding?.tmuxPane || binding.tmuxSession !== sessionName @@ -802,15 +815,16 @@ sessionAdminRoutes.get("/:sessionName/generation-record", terminalAuthGuard(), a // Cap the served suffix — the caller polls; a runaway record must not become a runaway response. const MAX_SUFFIX_BYTES = 8 * 1024 * 1024; try { - const fd = fs.openSync(transcriptPath, "r"); + if (expected) { original = await resolveExpectedConversation(expected, registry, tmuxAdapter, store); transcriptPath = original.path; } + const fd = fs.openSync(original?.path ?? transcriptPath, "r"); try { const stat = fs.fstatSync(fd); if (!stat.isFile()) throw new Error("record is not a regular file"); const totalBytes = stat.size; // Identity and bytes come from the same open file. Appends preserve this identity; // replacing a rollout at the same path changes it, even if its native id is unchanged. - const generationId = JSON.stringify([occupant?.generationUuid ?? null, sessionId, transcriptPath, stat.dev, stat.ino, stat.birthtimeMs]); - if (runtime === "codex") { + const generationId = JSON.stringify([original?.generation ?? occupant?.generationUuid ?? null, sessionId, transcriptPath, stat.dev, stat.ino, stat.birthtimeMs]); + if (runtime === "codex" || expected) { let header = ""; for (let offset = 0; offset < Math.min(totalBytes, MAX_SUFFIX_BYTES) && !header.includes("\n");) { const chunk = Buffer.alloc(Math.min(64 * 1024, totalBytes - offset, MAX_SUFFIX_BYTES - offset)); @@ -821,20 +835,33 @@ sessionAdminRoutes.get("/:sessionName/generation-record", terminalAuthGuard(), a } let meta; try { meta = JSON.parse(header.slice(0, header.indexOf("\n"))); } catch { /* explicit no-answer below */ } - if (meta?.type !== "session_meta" || meta.payload?.id !== sessionId) { + if (runtime === "claude-code" && expected) { + // Claude may prefix its JSONL with snapshots; identity must be present in the opened record. + const head = Buffer.alloc(Math.min(totalBytes, MAX_SUFFIX_BYTES)); + const n = fs.readSync(fd, head, 0, head.length, 0); + const ids = new Set(); + for (const line of head.subarray(0, n).toString("utf8").split("\n")) { + try { const row = JSON.parse(line); if (typeof row.sessionId === "string") ids.add(row.sessionId); } catch { /* incomplete last line */ } + } + if (ids.size !== 1 || !ids.has(sessionId!)) return c.json({ error: "expected_identity_mismatch", message: "The opened record does not identify only the expected original conversation." }, 409); + } else if (meta?.type !== "session_meta" || meta.payload?.id !== sessionId) { return c.json({ error: "record_identity_mismatch", message: `The rollout header does not identify current thread '${sessionId}'.` }, 409); } } - if (sinceRaw === undefined) return c.json({ generationId, sessionId, runtime, totalBytes }); + original?.assertCurrent(); + const receipt = expected ? { checkedIdentity: expected, identityContract: 1 as const } : {}; + if (sinceRaw === undefined) return c.json({ generationId, sessionId, runtime, totalBytes, ...receipt }); if (since > totalBytes) return c.json({ error: "record_truncated", message: "Generation record shrank below the requested byte boundary." }, 409); const length = Math.min(totalBytes - since, MAX_SUFFIX_BYTES); const buf = Buffer.alloc(length); const read = fs.readSync(fd, buf, 0, length, since); - return c.json({ generationId, sessionId, runtime, totalBytes, suffix: buf.subarray(0, read).toString("utf8"), truncated: totalBytes - since > read }); + return c.json({ generationId, sessionId, runtime, totalBytes, ...receipt, suffix: buf.subarray(0, read).toString("utf8"), truncated: totalBytes - since > read }); } finally { fs.closeSync(fd); } } catch (err) { - return c.json({ error: "record_unreadable", message: `Generation record for '${sessionName}' is unreadable: ${(err as Error).message}` }, 409); + return c.json({ error: expected ? (err as { code?: string }).code ?? "expected_identity_unavailable" : "record_unreadable", message: `Generation record for '${sessionName}' is unreadable: ${(err as Error).message}` }, 409); } -}); +}; +sessionAdminRoutes.get("/:sessionName/generation-record", terminalAuthGuard(), generationRecord); +sessionAdminRoutes.get("/:sessionName/generation-record-expected", terminalAuthGuard(), generationRecord); diff --git a/packages/daemon/src/routes/transport.ts b/packages/daemon/src/routes/transport.ts index 76a197bdf..45c89c055 100644 --- a/packages/daemon/src/routes/transport.ts +++ b/packages/daemon/src/routes/transport.ts @@ -1,4 +1,5 @@ -import { Hono } from "hono"; +import { expectedConversation, type ExpectedConversation } from "../domain/expected-conversation.js"; +import { Hono, type Context } from "hono"; import { CAPTURE_INPUT_BOX_DEADLINE_MS, CAPTURE_READABLE_BUDGET_MS, type SessionTransport, type TargetSpec } from "../domain/session-transport.js"; import { authBearerTokenMiddleware } from "../middleware/auth-bearer-token.js"; import { requireSenderIdentity, resolveRecordedIdentity, transportSenderSession } from "./require-sender-identity.js"; @@ -28,11 +29,12 @@ export function transportRoutes(opts?: { bearerToken?: string | null }): Hono { const terminalToken = opts?.bearerToken ?? null; router.use("*", authBearerTokenMiddleware({ expectedToken: terminalToken })); - router.post("/send", async (c) => { + const send = async (c: Context) => { const transport = c.get("sessionTransport" as never) as SessionTransport; const body = await c.req.json<{ session?: string; deliveryId?: string; + expectedIdentity?: unknown; text: string; verify?: boolean; force?: boolean; @@ -49,6 +51,12 @@ export function transportRoutes(opts?: { bearerToken?: string | null }): Hono { urgent?: boolean; }>(); + let expectedIdentity: ExpectedConversation | undefined; + if (c.req.path.endsWith("/send-expected") || body.expectedIdentity !== undefined) { + try { expectedIdentity = expectedConversation(body.expectedIdentity); } + catch (error) { return c.json({ ok: false, sent: false, reason: "invalid_expected_identity", error: (error as Error).message }, 400); } + } + // submitOnly (mechanics-gate fix d9b3989a) sends NO text — the Enter-only retry for staged // content; every other send still requires text. if (!body.session || (!body.text && !body.submitOnly)) { @@ -130,6 +138,7 @@ export function transportRoutes(opts?: { bearerToken?: string | null }): Hono { const result = await transport.send(body.session, body.text ?? "", { deliveryId: body.deliveryId, + expectedIdentity, verify: body.verify, force: body.force, waitForIdleMs: body.waitForIdleMs, @@ -158,6 +167,9 @@ export function transportRoutes(opts?: { bearerToken?: string | null }): Hono { if (!result.ok) { const statusMap: Record = { + expected_identity_mismatch: 409, + expected_identity_unavailable: 409, + invalid_expected_identity: 400, typing_guard_enabled: 409, caller_cancelled: 409, guard_target_unknown: 409, @@ -239,7 +251,10 @@ export function transportRoutes(opts?: { bearerToken?: string | null }): Hono { } return c.json(result); - }); + }; + router.post("/send", send); + // Separate capability path: an older daemon refuses with 404 instead of ignoring a new field. + router.post("/send-expected", send); router.post("/capture", async (c) => { const transport = c.get("sessionTransport" as never) as SessionTransport; diff --git a/packages/daemon/src/startup.ts b/packages/daemon/src/startup.ts index e251e2396..d7d9cafaf 100644 --- a/packages/daemon/src/startup.ts +++ b/packages/daemon/src/startup.ts @@ -1523,6 +1523,7 @@ export async function createDaemon(opts?: DaemonOptions): Promise transcriptStore, sessionTransport: (() => { const t = new SessionTransport({ + contextUsageStore, db, rigRepo, sessionRegistry,