diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8f8fd38405..b685a6df84 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -188,10 +188,36 @@ jobs: working-directory: ${{ matrix.package }} run: ${{ matrix.command }} + owned-session-contract-windows: + name: Caller-owned session contract (Windows named pipe) + needs: trust + if: needs.trust.outputs.allowed == 'true' + runs-on: windows-latest + timeout-minutes: 10 + steps: + - name: Checkout + uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + ref: ${{ inputs.checkout_ref || github.ref }} + persist-credentials: false + + - name: Setup Node.js + uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0 + with: + node-version: 22 + cache: npm + + - name: Install dependencies + run: npm ci + + - name: Test named-pipe owned cleanup + working-directory: packages/coding-agent + run: npx tsx ../../node_modules/vitest/dist/cli.js --run test/daemon-supervisor-process.test.ts -t "proves exact owned cleanup over a Windows named pipe" + build-check-test: name: build-check-test if: always() && (needs.trust.outputs.allowed == 'true' || inputs.require_trusted) - needs: [trust, build-check, test] + needs: [trust, build-check, test, owned-session-contract-windows] runs-on: ubuntu-latest steps: - name: Verify CI results @@ -199,7 +225,9 @@ jobs: TRUST_ALLOWED: ${{ needs.trust.outputs.allowed }} BUILD_CHECK_RESULT: ${{ needs.build-check.result }} TEST_RESULT: ${{ needs.test.result }} + WINDOWS_OWNED_SESSION_RESULT: ${{ needs.owned-session-contract-windows.result }} run: | test "$TRUST_ALLOWED" = true test "$BUILD_CHECK_RESULT" = success test "$TEST_RESULT" = success + test "$WINDOWS_OWNED_SESSION_RESULT" = success diff --git a/.pylon/features.yaml b/.pylon/features.yaml index 4276866092..587ba1a263 100644 --- a/.pylon/features.yaml +++ b/.pylon/features.yaml @@ -351,3 +351,19 @@ decisions: revisit_when: - Prime records a spawn-only parent session id in the session header or exposes an equivalent accessor on the extension-visible session manager. - The proxy-side parent-tree cancellation registry in rynfar/meridian#902 stops needing the parent edge inside the provider payload. + + caller-owned-session-environment-cleanup: + area: runtime-reliability + state: shipped + owner: shared + decision: retain + pylon_refs: + - https://github.com/pylon-code/prime-agent/issues/33 + - https://github.com/pylon-code/pylon/issues/199 + upstream_refs: + - https://github.com/PrimeIntellect-ai/prime-agent/tree/a903d4b6768f484bd6d459b7b0aa7dee38e461e2 + fork_change: caller-owned-session-environment-cleanup-v1 + upstream_support: Prime upstream does not expose a frozen caller-owned exact-environment contract, secret-free current-attachment proof, or bounded structured cleanup result that survives worker and supervisor replacement without weakening owner isolation. + revisit_when: + - Prime upstream exposes an equivalent public-root token, exact caller environment reuse, generation-scoped post-attach proof, and structured bounded cleanup outcomes. + - Pylon can remove the fork token without weakening native multi-instance isolation or its pre-create ACP fallback. diff --git a/.pylon/upstream-review.md b/.pylon/upstream-review.md index 607193dec4..ba24b06fbe 100644 --- a/.pylon/upstream-review.md +++ b/.pylon/upstream-review.md @@ -172,3 +172,12 @@ This ledger records Prime upstream evidence and the decision taken for each over - Deliberate boundaries: fork, clone, and branch do not carry the linkage, because the result is an independent session with its own provider key rather than a live child of a running parent. Recording the parent id no longer depends on the parent being persisted, which also aligns the inline and `AgentSessionRuntime` hosts with the daemon host's unconditional `newSession` call and its correct child `rlmDepth`. - Additive only: no daemon command, event, or response shape changes. The parent edge rides inside the `metadata.user_id` envelope extensions already produce. - Validation: `npm run check` clean. `test/suite/regressions/34-parent-session-identity.test.ts` passes 5/5 and 4 of its 5 cases fail when the inline path stops recording the parent id. Adjacent suites pass: session-manager unit suites, SDK session manager, session flush and git state, migrations, agent traces, context tree, RLM ledger, saved-session catalog, agent-session recursion, child provider identity, subagent runtime host, subagent model selection, subagent terminal messages, side questions, fast-mode children, agent-session runtime, suite compaction, concurrent sessions, daemon agent connection, daemon session id, daemon lazy subagents, daemon session list, ACP RLM subagents, RLM subagent display, clone command, session cwd, and agents-view state — 875 passes across 42 files. `test/extensions-runner.test.ts` fails 21 of 28 identically on unmodified `origin/pylon`, an environment-level extension-loading failure unrelated to this change. + +## 2026-08-31 — caller-owned session environment and cleanup contract candidate + +- Prime issue [#33](https://github.com/pylon-code/prime-agent/issues/33) freezes `caller_owned_session_environment_cleanup_v1` for the Pylon [#199](https://github.com/pylon-code/pylon/issues/199) native multi-instance gate. The public-root SDK token plus the daemon hello offers for the contract and authoritative cleanup form the preflight that selects native before create. Native attach must then return a current successful proof; a missing post-attach proof triggers bounded native cleanup and fails closed rather than pretending ACP cleaned a partial native create. +- Opted-in client-owned workers use one validated caller snapshot as the exact launch environment across create, attach fallback, worker recovery, and supervisor replacement. Legacy callers keep ambient merge behavior and never advertise the contract. Durable worker state stores only a non-secret contract marker, never the snapshot or an environment identity. +- Public proof is limited to protocol/app/build and supervisor/transport generations. Observable cleanup is single-flight and bounded, with fixed completed, already-completed, replacement-settled, owner-mismatch, uncertain, transport-failure, and unsupported outcomes. Arbitrary owners still cannot attach to or complete another owner's worker. +- The additive wire change keeps protocol version 7 and advances the schema revision from 28 to 29. Mixed-version clients remain usable through the legacy path when opt-in is absent. +- Local validation covers protocol and environment validation, old/new offer and echo compatibility, two-owner isolation, worker recovery, supervisor replacement, proof invalidation/republishing, wrong-owner denial, descriptor/log redaction, and cleanup. The Windows named-pipe case is committed but must run on the hosted `windows-latest` gate before merge. +- Cross-repository order remains Prime #33 and a reproducible package artifact first, then Pylon #199 consuming the exact proof. Revisit when upstream provides an equivalent frozen contract and Pylon can remove the fork token without weakening its ACP fallback. diff --git a/packages/coding-agent/.changes/33-caller-owned-session-contract.md b/packages/coding-agent/.changes/33-caller-owned-session-contract.md new file mode 100644 index 0000000000..800921543f --- /dev/null +++ b/packages/coding-agent/.changes/33-caller-owned-session-contract.md @@ -0,0 +1,2 @@ +- Added a negotiated caller-owned session environment proof and observable bounded cleanup outcomes for daemon SDK hosts. ([#33](https://github.com/pylon-code/prime-agent/issues/33)) +- Fixed Windows daemon startup so snapshot cache validation does not treat synthesized filesystem mode bits as POSIX permissions. ([#33](https://github.com/pylon-code/prime-agent/issues/33)) diff --git a/packages/coding-agent/docs/sdk.md b/packages/coding-agent/docs/sdk.md index 6800a59e31..93422cc2f3 100644 --- a/packages/coding-agent/docs/sdk.md +++ b/packages/coding-agent/docs/sdk.md @@ -77,6 +77,24 @@ if (!connection.supportsNegotiatedCapability("correlated_prompt_lifecycle_v1")) `supportsNegotiatedCapability()` is false before attach, while a new attach or reattach is pending, after transport or attachment invalidation, and after disposal. It becomes true only after the same physical transport returns a validated client capability echo and the exact snapshot commit succeeds. `supportsCorrelatedPromptLifecycle()` remains server-offer evidence used to construct the attach request. It is not negotiation proof. Do not substitute a hello offer, method presence, attach success, or package version for the post-attach accessor. Correlated runtime frames are withheld until the attach-side echo commits and are discarded when the echo omits the capability. Pre-proof retention is bounded by both frame count and conservative cumulative structural weight; overflow fails the adapter closed without retaining or reporting attributed payload content. A chunked replacement uses the same count and weight bounds for frames held behind its atomic snapshot fence. New same-connection attachment admission, attachment-epoch change, transport loss, disposal, or matching session close retires that old fence before any later proof can publish; delayed old snapshot frames are ignored until a fresh attachment commits. +### Caller-owned daemon session environments and cleanup + +Native multi-instance hosts must gate caller-owned daemon sessions with the exact `caller_owned_session_environment_cleanup_v1` contract. Require all three proofs: + +1. `PRIME_AGENT_SDK_FEATURES` from the package root includes the token. +2. The connected daemon hello offers both the same token and `authoritative_owned_session_cleanup_v1`. +3. `connection.getOwnedSessionContractProof()` returns the current post-attach proof. + +If either package-root or daemon-offer preflight fails, select ACP fallback before sending the native client-owned `create`. ACP fallback never cleans up a partially created native session. If the later attach does not return the current proof, fail native admission, run bounded native cleanup, and report that failure; do not treat ACP as retroactive cleanup for that partial native create. Do not infer support from a package version, method presence, constructor shape, protocol version, schema revision, or daemon offer alone. + +Pass one defensively captured `ownedSessionLaunchEnv` and a fresh `ownedSessionRecoveryConfig` to `DaemonAgentConnection`. Recovery config is required whenever the exact environment option is present; omitting it fails synchronously before attach. Use that same environment snapshot and `launchEnvMode: "replace"` on the preceding client-owned `create` request. The SDK validates and clones the option synchronously. A negotiated daemon uses the snapshot as the worker's whole caller-owned environment, adds only Prime-owned worker bootstrap variables, and reuses it with the recovery config for attach fallback and recovery. Without the exact options and negotiated token, legacy environment merging remains unchanged and the connection does not advertise this contract. + +The contract proof contains only the feature/status, protocol identity, schema revision, app/build identity, supervisor generation, and SDK-local transport generation. It is absent before attach, during reattach, after transport or attachment invalidation, after promotion/disposal, and for unproved peers. It never contains the environment, home, path, secret, hash, fingerprint, PID, socket, owner token, or canary. + +Use `disposeOwnedSession({ timeoutMs })` when cleanup must be observable. Concurrent calls join one operation, and `timeoutMs` is one strict total deadline for reconnection, authoritative queries, completion, side-question aborts, and unsupported-peer finalization. Its fixed statuses are `completed`, `already_completed`, `replacement_settled`, `owner_mismatch`, `uncertain`, `transport_failure`, and `unsupported`; `uncertain` also reports whether the last authoritative state was `active` or `stopping`. `replacement_settled` means an authenticated supervisor with a different generation answered the read-only cleanup query with `settled` for the connection's previously proved opaque route. Cleanup never sends completion on a replacement or pending route without a current internal attach proof. Each strict cleanup request is transport-bound and is never replayed after reconnect. Public attachment proof remains absent throughout disposal. Results never return raw errors or environment identity. `dispose()` remains the legacy best-effort `Promise` API. + +`DaemonClient.request()` keeps legacy reconnect replay by default. Pass `{ recoverAcrossReconnect: false }` only when one request must fail on transport close instead of crossing to a new daemon transport. + `DaemonClient.close()` is terminal owner disposal. `isClosed` becomes true, later `connect()` calls reject, and a live `DaemonAgentConnection` emits one terminal close. Normal and update recovery stop before any later restart, connect, attach, or restored-session query; already-running recovery callbacks are not cancellable but their results are discarded. `DaemonClientOptions.maxInboundFrameBytes` is the maximum raw bytes before LF in one inbound JSONL frame. It defaults to `DEFAULT_DAEMON_CLIENT_MAX_INBOUND_FRAME_BYTES` (128 MiB) and must be a positive safe integer. LF is excluded. A CR immediately before LF is counted and then stripped. diff --git a/packages/coding-agent/src/index.ts b/packages/coding-agent/src/index.ts index 127ae91752..c6e4795bc4 100644 --- a/packages/coding-agent/src/index.ts +++ b/packages/coding-agent/src/index.ts @@ -293,6 +293,11 @@ export { type AgentConnectionSlashCommand, type AgentConnectionState, DaemonAgentConnection, + type DaemonAgentConnectionOptions, + type DaemonOwnedSessionContractProof, + type DaemonOwnedSessionDaemonIdentity, + type DaemonOwnedSessionDisposeOptions, + type DaemonOwnedSessionDisposeResult, type ExpiredPromptLifecycle, InProcessAgentConnection, type PromptEventAttribution, @@ -318,6 +323,7 @@ export { type DaemonClientId, type DaemonClientMessageListener, type DaemonClientOptions, + type DaemonClientRequestOptions, type DaemonCommand, type DaemonCommandEnvelope, type DaemonCommandId, @@ -330,6 +336,8 @@ export { type DaemonOutbound, type DaemonOwnedSessionCleanupResult, type DaemonOwnedSessionCleanupStatus, + type DaemonOwnedSessionCompletionResult, + type DaemonOwnedSessionCompletionStatus, type DaemonProtocolInfo, type DaemonProtocolName, type DaemonProtocolVersion, @@ -413,7 +421,11 @@ export { Theme, type ThemeColor, } from "./modes/interactive/theme/theme.js"; -export { PRIME_AGENT_SDK_FEATURES, type PrimeAgentSdkFeature } from "./sdk-features.js"; +export { + CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE, + PRIME_AGENT_SDK_FEATURES, + type PrimeAgentSdkFeature, +} from "./sdk-features.js"; // Clipboard utilities export { copyToClipboard } from "./utils/clipboard.js"; export { parseFrontmatter, stripFrontmatter } from "./utils/frontmatter.js"; diff --git a/packages/coding-agent/src/main.ts b/packages/coding-agent/src/main.ts index 8740776ec5..d5c607f4e1 100644 --- a/packages/coding-agent/src/main.ts +++ b/packages/coding-agent/src/main.ts @@ -75,7 +75,11 @@ import { printTimings, resetTimings, time } from "./core/timings.js"; import { runMigrations, showDeprecationWarnings } from "./migrations.js"; import { isDaemonCatalogProcess, runDaemonCatalogProcess } from "./modes/daemon/daemon-catalog-process.js"; import { deserializeDaemonError } from "./modes/daemon/daemon-errors.js"; -import { collectDaemonClientEnv, collectDaemonLaunchEnv } from "./modes/daemon/daemon-protocol.js"; +import { + cloneCallerOwnedSessionLaunchEnv, + collectDaemonClientEnv, + collectDaemonLaunchEnv, +} from "./modes/daemon/daemon-protocol.js"; import { DAEMON_WORKER_ACTIVE_SESSION_ID_ENV, isDaemonWorkerProcess, @@ -111,6 +115,7 @@ import { ExtensionSelectorComponent } from "./modes/interactive/components/exten import { shouldRunOnboarding } from "./modes/interactive/onboarding.js"; import { initTheme, preloadCodeHighlighter, stopThemeWatcher } from "./modes/interactive/theme/theme.js"; import { handleConfigCommand } from "./package-manager-cli.js"; +import { CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE } from "./sdk-features.js"; import { isLocalPath } from "./utils/paths.js"; /** @@ -956,17 +961,28 @@ async function createDaemonClientConnection(options: { noSession?: boolean; supportsExtensionUi?: boolean; }): Promise<{ connection: DaemonAgentConnection; summary: SessionSummary }> { + // Capture before the first await so attach and every later recovery use one caller-owned snapshot. + const callerOwnedLaunchEnv = + options.clientOwned && options.activeSessionId === undefined + ? cloneCallerOwnedSessionLaunchEnv(collectDaemonLaunchEnv()) + : undefined; // Caller must have awaited ensureInteractiveDaemonRunning for this socket. const client = new DaemonClient(options.socketPath); await client.connect(); try { + await client.waitForHello(); + const supportsCallerOwnedEnvironmentContract = + callerOwnedLaunchEnv !== undefined && + client.supportsServerCapability(CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE) && + client.supportsServerCapability("authoritative_owned_session_cleanup_v1"); const attach = async (summary: SessionSummary) => { const connection = await DaemonAgentConnection.attach(client, getDaemonSummaryActiveSessionId(summary), { closeClientOnDispose: true, sendClientEnv: true, ownedSession: options.clientOwned, ownedSessionRecoveryConfig: options.clientOwned ? options.config : undefined, + ownedSessionLaunchEnv: supportsCallerOwnedEnvironmentContract ? callerOwnedLaunchEnv : undefined, supportsExtensionUi: options.supportsExtensionUi, recoverDaemon: () => ensureInteractiveDaemonRunning(options.socketPath), telemetryDisabled: options.config.telemetryDisabled, @@ -989,7 +1005,6 @@ async function createDaemonClientConnection(options: { } } if (options.clientOwned) { - await client.waitForHello(); if (!client.supportsServerCapability("client_owned_sessions")) { throw new DaemonCapabilityUnavailableError("create", "client_owned_sessions"); } @@ -1003,7 +1018,10 @@ async function createDaemonClientConnection(options: { noSession: options.noSession, env: collectDaemonClientEnv(), lifecycle: options.clientOwned ? "client_owned" : "resident", - launchEnv: collectDaemonLaunchEnv(), + launchEnv: supportsCallerOwnedEnvironmentContract + ? (callerOwnedLaunchEnv as Record) + : collectDaemonLaunchEnv(), + launchEnvMode: supportsCallerOwnedEnvironmentContract ? "replace" : undefined, }); if (!response.success) { throw deserializeDaemonError(response); diff --git a/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts b/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts index 45b23fa1cd..2a0aeb1cfa 100644 --- a/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts +++ b/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts @@ -32,6 +32,7 @@ import type { RefinementResult } from "../../core/refinement/index.js"; import type { DeleteSessionFileResult } from "../../core/session-file-actions.js"; import { SessionAlreadyActiveError } from "../../core/session-lease.js"; import type { SessionStats } from "../../core/session-stats.js"; +import { CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE, PRIME_AGENT_SDK_FEATURES } from "../../sdk-features.js"; import { DaemonCapabilityUnavailableError, type DaemonClient, @@ -39,8 +40,11 @@ import { } from "../daemon/daemon-client.js"; import { deserializeDaemonError } from "../daemon/daemon-errors.js"; import { + cloneCallerOwnedSessionLaunchEnv, collectDaemonClientEnv, collectDaemonLaunchEnv, + DAEMON_PROTOCOL_NAME, + DAEMON_PROTOCOL_VERSION, DAEMON_SNAPSHOT_GENERATION_NONCE_MIN_SCHEMA_REVISION, DAEMON_SUPPORTED_CLIENT_CAPABILITIES, type DaemonAttachResult, @@ -48,7 +52,10 @@ import { type DaemonCommand, type DaemonEventCursor, type DaemonOutbound, + type DaemonOwnedSessionCleanupResult, + type DaemonOwnedSessionCompletionResult, type DaemonReplayInfo, + type DaemonResponse, type DaemonSessionClosedReason, type DaemonSessionSnapshot, isUnknownDaemonCommandError, @@ -196,12 +203,38 @@ const MAX_COMPLETED_SNAPSHOTS = 128; const PROMPT_LIFECYCLE_TERMINAL_RETENTION = 256; const PROMPT_LIFECYCLE_TOMBSTONE_RETENTION = 256; const OWNED_SESSION_DISPOSE_RECONNECT_WAIT_MS = 10_000; +const OWNED_SESSION_DISPOSE_POLL_MS = 50; const updateTransportReconnects = new WeakMap>(); function delay(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } +function safeDaemonAppVersion(value: unknown): string | undefined { + return typeof value === "string" && + value.length <= 64 && + /^\d+\.\d+\.\d+(?:-[0-9A-Za-z.-]+)?(?:\+[0-9A-Za-z.-]+)?$/.test(value) + ? value + : undefined; +} + +function safeDaemonBuildId(value: unknown, appVersion: string | undefined): string | undefined { + if (typeof value !== "string" || value.length > 128) return undefined; + if (appVersion && value === `release-${appVersion}`) return value; + if (/^pylon-build-g[0-9a-f]{12}-r[1-9]\d*$/.test(value)) return value; + return /^(?:v?\d+\.\d+\.\d+-\d+-g)?[0-9a-f]{7,40}(?:-dirty)?$/.test(value) ? value : undefined; +} + +function isSafeSupervisorGeneration(value: unknown): value is string { + return ( + typeof value === "string" && /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(value) + ); +} + +function isUnknownActiveSessionError(error: unknown): boolean { + return error instanceof Error && error.message.startsWith("Unknown active session"); +} + function formatErrorSentence(error: unknown): string { const message = (error instanceof Error ? error.message : String(error)).trim(); if (!message) { @@ -312,12 +345,99 @@ export interface DaemonAgentConnectionOptions { supportsExtensionUi?: boolean; /** Dispose the connection by stopping its hidden worker instead of detaching. */ ownedSession?: boolean; - /** Fresh runtime context used only if the owned worker must be relaunched. */ + /** + * Fresh runtime context used only if the owned worker must be relaunched. + * Required whenever `ownedSessionLaunchEnv` is supplied. + */ ownedSessionRecoveryConfig?: AgentSessionRuntimeConfig; + /** + * Immutable caller-owned worker environment used for every owned attach and recovery. + * Requires `ownedSessionRecoveryConfig`. The SDK clones it once and never exposes it + * through contract proof or cleanup results. + */ + ownedSessionLaunchEnv?: Readonly>; /** Require the target worker to have been created with telemetry disabled. */ telemetryDisabled?: true; } +export interface DaemonOwnedSessionDaemonIdentity { + readonly protocolName: typeof DAEMON_PROTOCOL_NAME; + readonly protocolVersion: number; + readonly schemaRevision: number; + readonly appVersion?: string; + readonly buildId?: string; + readonly supervisorGeneration: string; + readonly transportGeneration: number; +} + +export interface DaemonOwnedSessionContractProof { + readonly feature: typeof CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE; + readonly status: "attached"; + readonly daemon: DaemonOwnedSessionDaemonIdentity; +} + +interface DaemonOwnedSessionDisposeBase { + readonly feature: typeof CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE; + readonly started?: DaemonOwnedSessionContractProof; + readonly observed?: DaemonOwnedSessionDaemonIdentity; +} + +export type DaemonOwnedSessionDisposeResult = + | (DaemonOwnedSessionDisposeBase & { + readonly status: "completed" | "already_completed"; + readonly started: DaemonOwnedSessionContractProof; + readonly observed: DaemonOwnedSessionDaemonIdentity; + readonly daemonReplaced: boolean; + }) + | (DaemonOwnedSessionDisposeBase & { + readonly status: "replacement_settled"; + readonly started: DaemonOwnedSessionContractProof; + readonly observed: DaemonOwnedSessionDaemonIdentity; + readonly daemonReplaced: true; + }) + | (DaemonOwnedSessionDisposeBase & { readonly status: "owner_mismatch" }) + | (DaemonOwnedSessionDisposeBase & { + readonly status: "uncertain"; + readonly reason: "active" | "stopping"; + }) + | (DaemonOwnedSessionDisposeBase & { readonly status: "transport_failure" | "unsupported" }); + +export interface DaemonOwnedSessionDisposeOptions { + /** One total deadline for observation, side-question aborts, and finalization. */ + readonly timeoutMs?: number; +} + +interface DaemonOwnedSessionCleanupAttachProof { + readonly activeSessionId: string; + readonly supervisorGeneration: string; + readonly transportGeneration: number; +} + +class OwnedSessionCleanupDeadlineError extends Error { + constructor() { + super("Owned session cleanup deadline elapsed"); + this.name = "OwnedSessionCleanupDeadlineError"; + } +} + +class OwnedSessionCleanupTransportFenceError extends Error { + constructor() { + super("Owned session cleanup transport changed during the request"); + this.name = "OwnedSessionCleanupTransportFenceError"; + } +} + +function hasSameOwnedSessionDaemonTransport( + left: DaemonOwnedSessionDaemonIdentity, + right: DaemonOwnedSessionDaemonIdentity | undefined, +): boolean { + return ( + right !== undefined && + left.supervisorGeneration === right.supervisorGeneration && + left.transportGeneration === right.transportGeneration + ); +} + /** * AgentConnection adapter for the local daemon JSONL socket transport. * @@ -418,6 +538,18 @@ function isCorrelatedPromptRuntimeFrame(message: DaemonOutbound): boolean { } export class DaemonAgentConnection implements AgentConnection { + private readonly client: DaemonClient; + private activeSessionId: string; + private readonly options: DaemonAgentConnectionOptions; + private ownedSessionLaunchEnv: Readonly> | undefined; + private currentOwnedSessionContractProof: DaemonOwnedSessionContractProof | undefined; + private currentOwnedSessionContractProofActiveSessionId: string | undefined; + private lastOwnedSessionContractProof: DaemonOwnedSessionContractProof | undefined; + private lastOwnedSessionContractProofActiveSessionId: string | undefined; + private ownedSessionDisposePromise: Promise | undefined; + private ownedSessionDisposeResult: DaemonOwnedSessionDisposeResult | undefined; + private ownedSessionCleanupAttachProof: DaemonOwnedSessionCleanupAttachProof | undefined; + private ownedSessionDisposeDeadline: number | undefined; private readonly listeners = new Set(); private readonly unsubscribeDaemonMessages: () => void; private readonly unsubscribeDaemonClose: () => void; @@ -546,11 +678,19 @@ export class DaemonAgentConnection implements AgentConnection { }); } - constructor( - private readonly client: DaemonClient, - private activeSessionId: string, - private readonly options: DaemonAgentConnectionOptions = {}, - ) { + constructor(client: DaemonClient, activeSessionId: string, options: DaemonAgentConnectionOptions = {}) { + const { ownedSessionLaunchEnv, ...connectionOptions } = options; + if (ownedSessionLaunchEnv !== undefined && options.ownedSession !== true) { + throw new Error("ownedSessionLaunchEnv requires ownedSession"); + } + if (ownedSessionLaunchEnv !== undefined && options.ownedSessionRecoveryConfig === undefined) { + throw new Error("ownedSessionLaunchEnv requires ownedSessionRecoveryConfig"); + } + this.client = client; + this.activeSessionId = activeSessionId; + this.options = connectionOptions; + this.ownedSessionLaunchEnv = + ownedSessionLaunchEnv === undefined ? undefined : cloneCallerOwnedSessionLaunchEnv(ownedSessionLaunchEnv); if (options.recoverDaemon) { this.client.enableRequestRecovery(); } @@ -606,6 +746,46 @@ export class DaemonAgentConnection implements AgentConnection { }); } + private hasCallerOwnedSessionEnvironmentContract(): boolean { + return ( + this.options.ownedSession === true && + this.options.ownedSessionRecoveryConfig !== undefined && + this.ownedSessionLaunchEnv !== undefined && + PRIME_AGENT_SDK_FEATURES.includes(CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE) + ); + } + + private serverOffersCallerOwnedSessionEnvironmentContract(): boolean { + return ( + this.client.supportsServerCapability(CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE) && + this.client.supportsServerCapability("authoritative_owned_session_cleanup_v1") + ); + } + + private ownedSessionLaunchFields(): { + launchEnv?: Record; + launchEnvMode?: "replace"; + } { + if (!this.options.ownedSession) return {}; + if (this.ownedSessionLaunchEnv) { + return { + launchEnv: this.ownedSessionLaunchEnv as Record, + launchEnvMode: "replace", + }; + } + return { launchEnv: collectDaemonLaunchEnv() }; + } + + private ownedSessionRecoveryFields(): { + recoveryConfig?: AgentSessionRuntimeConfig; + } { + return this.options.ownedSession && + this.options.ownedSessionRecoveryConfig && + this.client.supportsServerCapability("owned_session_recovery_context") + ? { recoveryConfig: this.options.ownedSessionRecoveryConfig } + : {}; + } + static async attach( client: DaemonClient, activeSessionId: string, @@ -691,6 +871,9 @@ export class DaemonAgentConnection implements AgentConnection { "chunked_snapshot", ...(this.supportsCorrelatedPromptLifecycle() ? (["correlated_prompt_lifecycle_v1"] as const) : []), ...(this.options.ownedSession ? (["client_owned_sessions"] as const) : []), + ...(this.hasCallerOwnedSessionEnvironmentContract() && this.serverOffersCallerOwnedSessionEnvironmentContract() + ? ([CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE] as const) + : []), ]; const attachmentEpoch = this.advanceAttachmentEpoch(); this.reserveSharedAttachment(requestedActiveSessionId); @@ -714,12 +897,8 @@ export class DaemonAgentConnection implements AgentConnection { clientId: this.clientId, capabilities, env: this.options.sendClientEnv ? collectDaemonClientEnv() : undefined, - launchEnv: this.options.ownedSession ? collectDaemonLaunchEnv() : undefined, - ...(this.options.ownedSession && - this.options.ownedSessionRecoveryConfig && - this.client.supportsServerCapability("owned_session_recovery_context") - ? { recoveryConfig: this.options.ownedSessionRecoveryConfig } - : {}), + ...this.ownedSessionLaunchFields(), + ...this.ownedSessionRecoveryFields(), telemetryDisabled: this.options.telemetryDisabled, resumeCursor: resumeCursor === undefined @@ -802,7 +981,7 @@ export class DaemonAgentConnection implements AgentConnection { allowWhileDisposing, ); this.commitStagedSnapshot(staged); - if (!this.disposing) this.publishNegotiatedCapabilityProof(negotiatedCapabilities, transportGeneration); + this.publishNegotiatedCapabilityProof(negotiatedCapabilities, transportGeneration); } else { validateSummaryIdentity(result, nextActiveSessionId); this.assertAttachmentCommit( @@ -818,7 +997,7 @@ export class DaemonAgentConnection implements AgentConnection { this.attachedSessionFile = result.sessionFile; this.latestSnapshot = undefined; this.latestSnapshotIsFresh = false; - if (!this.disposing) this.publishNegotiatedCapabilityProof(negotiatedCapabilities, transportGeneration); + this.publishNegotiatedCapabilityProof(negotiatedCapabilities, transportGeneration); } this.captureDaemonLogPath(); this.updateReconnectFailed = false; @@ -964,6 +1143,20 @@ export class DaemonAgentConnection implements AgentConnection { return this.client.supportsServerCapability("acp_mcp_servers"); } + /** Current secret-free proof that this owned attachment uses the caller-owned environment contract. */ + getOwnedSessionContractProof(): DaemonOwnedSessionContractProof | undefined { + if ( + this.disposing || + this.disposed || + !this.currentOwnedSessionContractProof || + this.currentOwnedSessionContractProofActiveSessionId !== this.activeSessionId || + !this.supportsNegotiatedCapability(CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE) + ) { + return undefined; + } + return this.currentOwnedSessionContractProof; + } + /** * Whether the daemon proved this client capability for the current committed * attachment. This is false while attach/reattach is pending and after the @@ -1959,6 +2152,10 @@ export class DaemonAgentConnection implements AgentConnection { "chunked_snapshot", ...(this.supportsCorrelatedPromptLifecycle() ? (["correlated_prompt_lifecycle_v1"] as const) : []), ...(this.options.ownedSession ? (["client_owned_sessions"] as const) : []), + ...(this.hasCallerOwnedSessionEnvironmentContract() && + this.client.supportsServerCapability(CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE) + ? ([CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE] as const) + : []), ]; let result!: DaemonAttachResult; let snapshot!: DaemonSessionSnapshot; @@ -1980,7 +2177,8 @@ export class DaemonAgentConnection implements AgentConnection { clientId: this.clientId, capabilities, env: this.options.sendClientEnv ? collectDaemonClientEnv() : undefined, - launchEnv: this.options.ownedSession ? collectDaemonLaunchEnv() : undefined, + ...this.ownedSessionLaunchFields(), + ...this.ownedSessionRecoveryFields(), telemetryDisabled: this.options.telemetryDisabled, } : { @@ -1991,6 +2189,8 @@ export class DaemonAgentConnection implements AgentConnection { clientId: this.clientId, capabilities, env: this.options.sendClientEnv ? collectDaemonClientEnv() : undefined, + ...this.ownedSessionLaunchFields(), + ...this.ownedSessionRecoveryFields(), telemetryDisabled: this.options.telemetryDisabled, }, ); @@ -2198,10 +2398,204 @@ export class DaemonAgentConnection implements AgentConnection { }; } - async dispose(): Promise { + async disposeOwnedSession(options: DaemonOwnedSessionDisposeOptions = {}): Promise { + if (this.ownedSessionDisposeResult) return this.ownedSessionDisposeResult; + if (this.ownedSessionDisposePromise) return this.ownedSessionDisposePromise; + const timeoutMs = options.timeoutMs ?? OWNED_SESSION_DISPOSE_RECONNECT_WAIT_MS; + if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0) { + throw new RangeError("timeoutMs must be a positive safe integer"); + } + const deadline = Date.now() + timeoutMs; + const operation = this.disposeOwnedSessionUnserialized(deadline).then((result) => { + this.ownedSessionDisposeResult = result; + return result; + }); + this.ownedSessionDisposePromise = operation; + return operation; + } + + private async disposeOwnedSessionUnserialized(deadline: number): Promise { + const feature = CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE; + const started = this.lastOwnedSessionContractProof; + const startedActiveSessionId = this.lastOwnedSessionContractProofActiveSessionId; if (this.disposed || this.disposing) { + return Object.freeze({ feature, status: "unsupported", ...(started ? { started } : {}) }); + } + this.ownedSessionDisposeDeadline = deadline; + const serverRoutedActiveSessionId = this.serverRoutedActiveSessionId(); + if ( + this.currentOwnedSessionContractProof && + this.currentOwnedSessionContractProofActiveSessionId === serverRoutedActiveSessionId + ) { + this.ownedSessionCleanupAttachProof = { + activeSessionId: serverRoutedActiveSessionId, + supervisorGeneration: this.currentOwnedSessionContractProof.daemon.supervisorGeneration, + transportGeneration: this.currentOwnedSessionContractProof.daemon.transportGeneration, + }; + } + this.disposing = true; + this.advanceAttachmentEpoch(); + let result: DaemonOwnedSessionDisposeResult; + if ( + !this.hasCallerOwnedSessionEnvironmentContract() || + !started || + startedActiveSessionId !== serverRoutedActiveSessionId + ) { + result = Object.freeze({ feature, status: "unsupported", ...(started ? { started } : {}) }); + const action = this.options.ownedSession + ? started && startedActiveSessionId !== serverRoutedActiveSessionId + ? "none" + : "complete" + : "detach"; + await this.finalizeDisposedConnection(serverRoutedActiveSessionId, action); + return result; + } + try { + result = await this.observeOwnedSessionCleanup(started, serverRoutedActiveSessionId, deadline); + } catch { + result = Object.freeze({ feature, status: "transport_failure", started }); + } + await this.finalizeDisposedConnection(serverRoutedActiveSessionId, "none"); + return result; + } + + private async observeOwnedSessionCleanup( + started: DaemonOwnedSessionContractProof, + activeSessionId: string, + deadline: number, + ): Promise { + const feature = CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE; + let lastStatus: DaemonOwnedSessionCleanupResult["status"] | undefined; + let lastObserved: DaemonOwnedSessionDaemonIdentity | undefined; + let completionAttempted = false; + while (Date.now() < deadline) { + if (!this.client.isConnected) { + const reconnect = this.reconnectPromise; + if (!reconnect) { + return Object.freeze({ + feature, + status: "transport_failure", + started, + ...(lastObserved ? { observed: lastObserved } : {}), + }); + } + await this.awaitOwnedSessionDisposeDeadline(reconnect.catch(() => undefined)).catch(() => undefined); + if (!this.client.isConnected) continue; + } + const observed = this.currentOwnedSessionDaemonIdentity(); + if (!observed) { + return Object.freeze({ feature, status: "unsupported", started }); + } + lastObserved = observed; + const cleanupAttachProof = this.ownedSessionCleanupAttachProof; + const hasCurrentAttachProof = + cleanupAttachProof?.activeSessionId === activeSessionId && + cleanupAttachProof.supervisorGeneration === observed.supervisorGeneration && + cleanupAttachProof.transportGeneration === observed.transportGeneration; + try { + const cleanup = await this.requestData({ + type: "get_owned_session_cleanup", + activeSessionId, + }); + if (!hasSameOwnedSessionDaemonTransport(observed, this.currentOwnedSessionDaemonIdentity())) { + throw new OwnedSessionCleanupTransportFenceError(); + } + if (cleanup.status !== "active" && cleanup.status !== "stopping" && cleanup.status !== "settled") { + return Object.freeze({ feature, status: "transport_failure", started, observed }); + } + lastStatus = cleanup.status; + const daemonReplaced = started.daemon.supervisorGeneration !== observed.supervisorGeneration; + if (cleanup.status === "settled") { + if (daemonReplaced) { + return Object.freeze({ + feature, + status: "replacement_settled", + started, + observed, + daemonReplaced: true, + }); + } + return Object.freeze({ + feature, + status: completionAttempted ? "completed" : "already_completed", + started, + observed, + daemonReplaced: false, + }); + } + if (cleanup.status === "active" && hasCurrentAttachProof) { + completionAttempted = true; + const completion = await this.completeOwnedSessionWithStructuredFailure(activeSessionId, observed); + if (completion === "owner_mismatch") { + return Object.freeze({ feature, status: "owner_mismatch", started, observed }); + } + if (completion === "completed") { + return Object.freeze({ + feature, + status: "completed", + started, + observed, + daemonReplaced, + }); + } + } + } catch (error) { + if (error instanceof OwnedSessionCleanupDeadlineError) break; + if (!this.client.isConnected && !this.reconnectPromise) { + return Object.freeze({ feature, status: "transport_failure", started, observed }); + } + } + const remainingMs = deadline - Date.now(); + if (remainingMs > 0) await delay(Math.min(OWNED_SESSION_DISPOSE_POLL_MS, remainingMs)); + } + if (lastStatus === "active" || lastStatus === "stopping") { + return Object.freeze({ + feature, + status: "uncertain", + reason: lastStatus, + started, + ...(lastObserved ? { observed: lastObserved } : {}), + }); + } + return Object.freeze({ + feature, + status: "transport_failure", + started, + ...(lastObserved ? { observed: lastObserved } : {}), + }); + } + + private async completeOwnedSessionWithStructuredFailure( + activeSessionId: string, + expectedDaemon: DaemonOwnedSessionDaemonIdentity, + ): Promise<"completed" | "owner_mismatch"> { + if (!hasSameOwnedSessionDaemonTransport(expectedDaemon, this.currentOwnedSessionDaemonIdentity())) { + throw new OwnedSessionCleanupTransportFenceError(); + } + const response = await this.requestDaemonCommandWithinOwnedSessionDeadline({ + type: "complete_owned_session", + activeSessionId, + }); + if (!hasSameOwnedSessionDaemonTransport(expectedDaemon, this.currentOwnedSessionDaemonIdentity())) { + throw new OwnedSessionCleanupTransportFenceError(); + } + if (!response.success) { + if (response.errorInfo?.code === "owned_session_owner_mismatch") return "owner_mismatch"; + const error = deserializeDaemonError(response); + this.definitiveRequestErrors.add(error); + throw error; + } + const completion = response.data as DaemonOwnedSessionCompletionResult | undefined; + if (completion === undefined || completion.status === "completed") return "completed"; + throw new Error("Daemon returned invalid owned session completion result"); + } + + async dispose(): Promise { + if (this.ownedSessionDisposePromise) { + await this.ownedSessionDisposePromise; return; } + if (this.disposed || this.disposing) return; this.disposing = true; this.advanceAttachmentEpoch(); if (this.options.ownedSession && !this.client.isConnected && this.reconnectPromise) { @@ -2209,26 +2603,42 @@ export class DaemonAgentConnection implements AgentConnection { () => undefined, ); } + await this.finalizeDisposedConnection( + this.serverRoutedActiveSessionId(), + this.options.ownedSession ? "complete" : "detach", + ); + } + + private serverRoutedActiveSessionId(): string { + return [...this.pendingReattachActiveSessionIds].at(-1) ?? this.activeSessionId; + } + + private async finalizeDisposedConnection( + serverRoutedActiveSessionId: string, + action: "complete" | "detach" | "none", + ): Promise { this.disposed = true; this.invalidateNegotiatedCapabilityProof(); this.updateRestartPending = false; await Promise.allSettled([...this.activeSideQuestionIds].map((id) => this.abortSideQuestion(id))); this.unsubscribeDaemonMessages(); this.unsubscribeDaemonClose(); - const pendingActiveSessionId = [...this.pendingReattachActiveSessionIds].at(-1); - const serverRoutedActiveSessionId = pendingActiveSessionId ?? this.activeSessionId; this.invalidateSharedAttachment(serverRoutedActiveSessionId); - if (this.options.ownedSession) { + if (action === "complete") { await this.requestOk({ type: "complete_owned_session", activeSessionId: serverRoutedActiveSessionId }).catch( () => undefined, ); - } else { + } else if (action === "detach") { await this.requestOk({ type: "detach", activeSessionId: serverRoutedActiveSessionId }).catch(() => undefined); } - if (this.options.closeClientOnDispose) { - this.client.close(); - } + if (this.options.closeClientOnDispose) this.client.close(); this.rejectSnapshotAssemblies(new Error("Daemon connection disposed during snapshot transfer")); + this.ownedSessionLaunchEnv = undefined; + this.currentOwnedSessionContractProof = undefined; + this.currentOwnedSessionContractProofActiveSessionId = undefined; + this.lastOwnedSessionContractProof = undefined; + this.lastOwnedSessionContractProofActiveSessionId = undefined; + this.ownedSessionCleanupAttachProof = undefined; } async promoteToResident(): Promise { @@ -2244,6 +2654,11 @@ export class DaemonAgentConnection implements AgentConnection { const result = await operation(promoteOwnedSession); if (promoteOwnedSession) { this.options.ownedSession = false; + this.ownedSessionLaunchEnv = undefined; + this.currentOwnedSessionContractProof = undefined; + this.currentOwnedSessionContractProofActiveSessionId = undefined; + this.lastOwnedSessionContractProof = undefined; + this.lastOwnedSessionContractProofActiveSessionId = undefined; } return result; }); @@ -2309,6 +2724,14 @@ export class DaemonAgentConnection implements AgentConnection { return; } catch (error) { lastError = error instanceof Error ? error : new Error(String(error)); + if ( + this.disposing && + this.options.ownedSession === true && + this.client.isConnected && + isUnknownActiveSessionError(lastError) + ) { + return; + } if (this.client.isClosed) { this.emitOwnerClosedTerminal(); return; @@ -2337,6 +2760,43 @@ export class DaemonAgentConnection implements AgentConnection { return this.reconnectPromise; } + private awaitOwnedSessionDisposeDeadline(operation: Promise): Promise { + const deadline = this.ownedSessionDisposeDeadline; + if (deadline === undefined) return operation; + const remainingMs = deadline - Date.now(); + if (remainingMs <= 0) return Promise.reject(new OwnedSessionCleanupDeadlineError()); + return new Promise((resolveOperation, rejectOperation) => { + const timer = setTimeout(() => rejectOperation(new OwnedSessionCleanupDeadlineError()), remainingMs); + operation.then( + (value) => { + clearTimeout(timer); + resolveOperation(value); + }, + (error) => { + clearTimeout(timer); + rejectOperation(error); + }, + ); + }); + } + + private requestDaemonCommandWithinOwnedSessionDeadline( + command: DaemonCommandBody, + timeoutMs?: number, + options?: Parameters[2], + ): Promise { + const deadline = this.ownedSessionDisposeDeadline; + let effectiveTimeoutMs = timeoutMs; + let effectiveOptions = options; + if (deadline !== undefined) { + const remainingMs = deadline - Date.now(); + if (remainingMs <= 0) return Promise.reject(new OwnedSessionCleanupDeadlineError()); + effectiveTimeoutMs = Math.max(1, Math.min(timeoutMs ?? remainingMs, remainingMs)); + effectiveOptions = { ...options, recoverAcrossReconnect: false }; + } + return this.awaitOwnedSessionDisposeDeadline(this.client.request(command, effectiveTimeoutMs, effectiveOptions)); + } + private async requestOk(command: DaemonCommandBody): Promise { await this.requestData(command); } @@ -2346,7 +2806,7 @@ export class DaemonAgentConnection implements AgentConnection { timeoutMs?: number, options?: Parameters[2], ): Promise { - const response = await this.client.request(command, timeoutMs, options); + const response = await this.requestDaemonCommandWithinOwnedSessionDeadline(command, timeoutMs, options); if (!response.success) { const error = deserializeDaemonError(response); this.definitiveRequestErrors.add(error); @@ -2987,6 +3447,8 @@ export class DaemonAgentConnection implements AgentConnection { private invalidateNegotiatedCapabilityProof(clearRuntimeCapabilities = true): void { this.attachmentInvalidationRevision++; + this.currentOwnedSessionContractProof = undefined; + this.currentOwnedSessionContractProofActiveSessionId = undefined; this.negotiatedCapabilities = new Set(); this.negotiatedTransportGeneration = undefined; if (clearRuntimeCapabilities) { @@ -3062,14 +3524,84 @@ export class DaemonAgentConnection implements AgentConnection { if (this.replacementReconciliationFailed) { throw new Error("Daemon connection replacement reconciliation has failed"); } + if (this.disposing) { + this.discardUnsolicitedRuntimeSnapshots = true; + this.negotiatedCapabilities = new Set(); + this.negotiatedTransportGeneration = undefined; + this.negotiatedRuntimeCapabilities = new Set(); + this.negotiatedRuntimeTransportGeneration = undefined; + this.currentOwnedSessionContractProof = undefined; + this.currentOwnedSessionContractProofActiveSessionId = undefined; + const daemon = capabilities.has(CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE) + ? this.currentOwnedSessionDaemonIdentity() + : undefined; + this.ownedSessionCleanupAttachProof = daemon + ? { + activeSessionId: this.activeSessionId, + supervisorGeneration: daemon.supervisorGeneration, + transportGeneration: daemon.transportGeneration, + } + : undefined; + this.pendingNegotiatedRuntimeFrames = []; + this.pendingNegotiatedRuntimeFrameWeight = 0; + return; + } this.discardUnsolicitedRuntimeSnapshots = false; this.negotiatedCapabilities = capabilities; this.negotiatedTransportGeneration = transportGeneration; this.negotiatedRuntimeCapabilities = capabilities; this.negotiatedRuntimeTransportGeneration = transportGeneration; + this.currentOwnedSessionContractProof = capabilities.has(CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE) + ? this.createOwnedSessionContractProof() + : undefined; + this.currentOwnedSessionContractProofActiveSessionId = this.currentOwnedSessionContractProof + ? this.activeSessionId + : undefined; + if (this.currentOwnedSessionContractProof) { + this.lastOwnedSessionContractProof = this.currentOwnedSessionContractProof; + this.lastOwnedSessionContractProofActiveSessionId = this.activeSessionId; + } this.releasePendingNegotiatedRuntimeFrames(); } + private currentOwnedSessionDaemonIdentity(): DaemonOwnedSessionDaemonIdentity | undefined { + const hello = this.client.hello; + if ( + !hello || + !this.hasCallerOwnedSessionEnvironmentContract() || + !this.serverOffersCallerOwnedSessionEnvironmentContract() || + hello.protocol.name !== DAEMON_PROTOCOL_NAME || + !Number.isSafeInteger(hello.protocol.version) || + hello.protocol.version < DAEMON_PROTOCOL_VERSION || + !Number.isSafeInteger(hello.schemaRevision) || + (hello.schemaRevision ?? 0) < 29 || + !isSafeSupervisorGeneration(hello.supervisorGeneration) + ) { + return undefined; + } + const appVersion = safeDaemonAppVersion(hello.appVersion); + const buildId = safeDaemonBuildId(hello.runtime?.buildId, appVersion); + return Object.freeze({ + protocolName: DAEMON_PROTOCOL_NAME, + protocolVersion: hello.protocol.version, + schemaRevision: hello.schemaRevision!, + ...(appVersion === undefined ? {} : { appVersion }), + ...(buildId === undefined ? {} : { buildId }), + supervisorGeneration: hello.supervisorGeneration, + transportGeneration: this.client.getTransportGeneration(), + }); + } + + private createOwnedSessionContractProof(): DaemonOwnedSessionContractProof | undefined { + const daemon = this.currentOwnedSessionDaemonIdentity(); + if (!daemon) return undefined; + return Object.freeze({ + feature: CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE, + status: "attached", + daemon, + }); + } + private negotiatedCapabilitiesFromAttach( result: DaemonAttachResult, requestedCapabilities: readonly DaemonClientCapability[], diff --git a/packages/coding-agent/src/modes/agent-connection/index.ts b/packages/coding-agent/src/modes/agent-connection/index.ts index 454945c54e..69d24bff3a 100644 --- a/packages/coding-agent/src/modes/agent-connection/index.ts +++ b/packages/coding-agent/src/modes/agent-connection/index.ts @@ -8,6 +8,13 @@ export type { PromptLifecycleSnapshot, PromptLifecycleStateSnapshot, } from "../../core/prompt-lifecycle.js"; +export type { + DaemonAgentConnectionOptions, + DaemonOwnedSessionContractProof, + DaemonOwnedSessionDaemonIdentity, + DaemonOwnedSessionDisposeOptions, + DaemonOwnedSessionDisposeResult, +} from "./daemon-agent-connection.js"; export { DaemonAgentConnection } from "./daemon-agent-connection.js"; export { InProcessAgentConnection } from "./in-process-agent-connection.js"; export { createAgentConnectionCommands, createAgentConnectionState } from "./snapshot.js"; diff --git a/packages/coding-agent/src/modes/daemon/daemon-client.ts b/packages/coding-agent/src/modes/daemon/daemon-client.ts index bbde1900c0..233289a350 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-client.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-client.ts @@ -50,6 +50,8 @@ export class DaemonInboundFrameTooLargeError extends Error { export interface DaemonClientRequestOptions { onProgress?: DaemonClientProgressListener; + /** Preserve and replay this request after reconnect. Defaults to true. */ + recoverAcrossReconnect?: boolean; } interface PendingDaemonRequest { @@ -62,6 +64,7 @@ interface PendingDaemonRequest { wireData: string; awaitingReconnect: boolean; acknowledgeResult: boolean; + recoverAcrossReconnect: boolean; /** Re-checked against the new hello before a reconnect replay. */ compatibilities: readonly DaemonCommandCompatibility[]; } @@ -456,6 +459,7 @@ export class DaemonClient { wireData, awaitingReconnect: false, acknowledgeResult, + recoverAcrossReconnect: options.recoverAcrossReconnect !== false, compatibilities, }; this.pendingRequests.set(id, pending); @@ -616,7 +620,7 @@ export class DaemonClient { private rejectAll(error: Error, preservePendingRequests = false): void { for (const [id, pending] of this.pendingRequests) { - if (preservePendingRequests) { + if (preservePendingRequests && pending.recoverAcrossReconnect) { if (pending.timeout) { clearTimeout(pending.timeout); pending.timeout = undefined; diff --git a/packages/coding-agent/src/modes/daemon/daemon-protocol.ts b/packages/coding-agent/src/modes/daemon/daemon-protocol.ts index a4e3e4d4eb..6e76cee18a 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-protocol.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-protocol.ts @@ -28,6 +28,7 @@ import type { import type { QueuedMessageLane, QueuedMessageMutation } from "../../core/session-action-store.js"; import type { SessionCwdIssue } from "../../core/session-cwd.js"; import type { DeleteSessionFileResult } from "../../core/session-file-actions.js"; +import { CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE } from "../../sdk-features.js"; import type { AgentConnectionAgentStatus, AgentConnectionHeartbeat, @@ -79,8 +80,9 @@ export const DAEMON_COMMAND_ENVELOPE_MIN_PROTOCOL_VERSION = 7; // Revision 26 correlates snapshot failures that occur before a begin frame can be emitted. // Revision 27 adds a capability-gated authoritative owned-session cleanup query. // Revision 28 negotiates fresh snapshot generations and adds private worker chunk routing metadata. -export const DAEMON_SCHEMA_REVISION = 28; -export const DAEMON_SCHEMA_ID = "protocol-7-schema-28-7c0c21a689b5"; +// Revision 29 capability-gates exact caller-owned launch environments and observable cleanup results. +export const DAEMON_SCHEMA_REVISION = 29; +export const DAEMON_SCHEMA_ID = "protocol-7-schema-29-5450eb231171"; export type DaemonProtocolName = typeof DAEMON_PROTOCOL_NAME; export type DaemonProtocolVersion = number; @@ -100,7 +102,8 @@ export type DaemonClientCapability = | "chunked_snapshot" | "immutable_snapshot_transfer_v1" | "client_owned_sessions" - | "correlated_prompt_lifecycle_v1"; + | "correlated_prompt_lifecycle_v1" + | typeof CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE; export type DaemonPromptAdmissionCancellationStatus = "cancelled" | "owned" | "unknown"; export interface DaemonPromptAdmissionCancellationResult { status: DaemonPromptAdmissionCancellationStatus; @@ -109,6 +112,11 @@ export type DaemonOwnedSessionCleanupStatus = "active" | "stopping" | "settled"; export interface DaemonOwnedSessionCleanupResult { status: DaemonOwnedSessionCleanupStatus; } + +export type DaemonOwnedSessionCompletionStatus = "completed"; +export interface DaemonOwnedSessionCompletionResult { + status: DaemonOwnedSessionCompletionStatus; +} export type DaemonServerCapability = | DaemonClientCapability | "delete_rlm_subagent" @@ -161,10 +169,13 @@ export const DAEMON_SUPPORTED_CLIENT_CAPABILITIES: readonly DaemonClientCapabili "immutable_snapshot_transfer_v1", "client_owned_sessions", "correlated_prompt_lifecycle_v1", + CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE, ]; export const DAEMON_DEFAULT_SERVER_CAPABILITIES: readonly DaemonServerCapability[] = [ - ...DAEMON_SUPPORTED_CLIENT_CAPABILITIES, + ...DAEMON_SUPPORTED_CLIENT_CAPABILITIES.filter( + (capability) => capability !== CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE, + ), "delete_rlm_subagent", "heartbeat_catalog", "heartbeat_management", @@ -187,6 +198,7 @@ export const DAEMON_DEFAULT_SERVER_CAPABILITIES: readonly DaemonServerCapability export const DAEMON_SUPERVISOR_SERVER_CAPABILITIES: readonly DaemonServerCapability[] = [ ...DAEMON_DEFAULT_SERVER_CAPABILITIES, "authoritative_owned_session_cleanup_v1", + CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE, ]; export interface DaemonRuntimeIdentity { @@ -224,8 +236,12 @@ export interface DaemonClientEnv { export type DaemonSessionLifecycle = "resident" | "client_owned"; +export type DaemonLaunchEnvMode = "replace"; + export interface DaemonLaunchEnv { launchEnv?: Record; + /** Capability-gated: use only this snapshot plus Prime-owned worker bootstrap variables. */ + launchEnvMode?: DaemonLaunchEnvMode; } /** @@ -256,13 +272,68 @@ export function collectDaemonClientEnv(source: NodeJS.ProcessEnv = process.env): export function collectDaemonLaunchEnv(source: NodeJS.ProcessEnv = process.env): Record { const env: Record = {}; for (const [key, value] of Object.entries(source)) { - if (value !== undefined && !key.startsWith("PRIME_AGENT_INTERNAL_")) { + const normalizedKey = key.toUpperCase(); + if (value !== undefined && !normalizedKey.startsWith("PRIME_AGENT_INTERNAL_") && normalizedKey !== "RLM_DEPTH") { env[key] = value; } } return env; } +/** Clone an untrusted caller-owned worker environment without retaining or reporting its values. */ +export function cloneCallerOwnedSessionLaunchEnv( + source: Readonly>, + platform: NodeJS.Platform = process.platform, +): Readonly> { + if (typeof source !== "object" || source === null) { + throw new TypeError("ownedSessionLaunchEnv must be a string environment record"); + } + let descriptors: PropertyDescriptorMap; + try { + if (Array.isArray(source)) { + throw new TypeError("ownedSessionLaunchEnv must be a string environment record"); + } + descriptors = Object.getOwnPropertyDescriptors(source); + } catch { + throw new TypeError("ownedSessionLaunchEnv must be an inspectable string environment record"); + } + const environment = Object.create(null) as Record; + const windowsKeys = new Set(); + for (const key of Reflect.ownKeys(descriptors)) { + if (typeof key !== "string") { + throw new TypeError("ownedSessionLaunchEnv must use string environment keys"); + } + const descriptor = descriptors[key]!; + const value = descriptor.value; + if ( + !("value" in descriptor) || + key.length === 0 || + key.includes("=") || + key.includes("\0") || + key.toUpperCase().startsWith("PRIME_AGENT_INTERNAL_") || + key.toUpperCase() === "RLM_DEPTH" || + typeof value !== "string" || + value.includes("\0") + ) { + throw new TypeError("ownedSessionLaunchEnv must be a string environment record without reserved keys"); + } + if (platform === "win32") { + const normalizedKey = key.toUpperCase(); + if (windowsKeys.has(normalizedKey)) { + throw new TypeError("ownedSessionLaunchEnv contains duplicate Windows environment keys"); + } + windowsKeys.add(normalizedKey); + } + Object.defineProperty(environment, key, { + value, + enumerable: true, + configurable: false, + writable: false, + }); + } + return Object.freeze(environment); +} + export interface DaemonReplayInfo { status: DaemonReplayStatus; fromSequence?: DaemonEventSequence; @@ -744,6 +815,11 @@ const AUTHORITATIVE_OWNED_SESSION_CLEANUP_COMMAND = { minSchemaRevision: 27, capability: "authoritative_owned_session_cleanup_v1", } as const; +const CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_COMMAND = { + minProtocol: 7, + minSchemaRevision: 29, + capability: CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE, +} as const; const DELETE_RLM_SUBAGENT_COMMAND = { minProtocol: 7, capability: "delete_rlm_subagent", @@ -898,6 +974,12 @@ export function getDaemonCommandCompatibilities(command: DaemonCommand): readonl if ((command.type === "attach" || command.type === "reattach") && command.recoveryConfig !== undefined) { requirements.push(OWNED_SESSION_RECOVERY_CONTEXT); } + if ( + (command.type === "create" || command.type === "attach" || command.type === "reattach") && + command.launchEnvMode === "replace" + ) { + requirements.push(CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_COMMAND); + } if (command.type === "attach" && command.snapshotGenerationNonce !== undefined) { requirements.push(SNAPSHOT_GENERATION_NONCE_COMMAND); } @@ -939,7 +1021,8 @@ export type DaemonErrorInfo = | { code: "missing_session_cwd"; issue: SessionCwdIssue } | { code: "session_import_file_not_found"; filePath: string } | { code: "session_already_active"; sessionPath: string; activeSessionId?: string } - | { code: "command_result_uncertain"; clientId: DaemonClientId; commandId: DaemonCommandId }; + | { code: "command_result_uncertain"; clientId: DaemonClientId; commandId: DaemonCommandId } + | { code: "owned_session_owner_mismatch" }; export type DaemonSessionClosedReason = "killed" | "shutdown" | "completed" | "replaced" | "update"; export type DaemonClosingReason = "shutdown" | "update"; diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index 1927471677..c2b2e6583f 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -4,6 +4,7 @@ import { chmodSync, lstatSync, mkdirSync, readdirSync, readFileSync, renameSync, import { createServer, type Server, type Socket } from "node:net"; import { dirname, join, resolve } from "node:path"; import { Writable } from "node:stream"; +import { isDeepStrictEqual } from "node:util"; import { getLogger } from "@earendil-works/pi-ai"; import { createCliSubprocessEnv, createCliSubprocessLaunchSpec } from "../../cli/subprocess-launch.js"; import { @@ -56,6 +57,7 @@ import { canonicalSessionPath, getProcessStartId, SessionAlreadyActiveError } fr import { getSessionArtifactPathForFile, readSessionInfo, type SessionInfo } from "../../core/session-manager.js"; import { looksLikeSessionPath } from "../../core/session-resolver.js"; import { SettingsManager } from "../../core/settings-manager.js"; +import { CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE } from "../../sdk-features.js"; import { isProcessAlive, processIdExists, signalProcessGroupOrProcess } from "../../utils/child-process.js"; import type { AgentConnectionHeartbeat } from "../agent-connection/types.js"; import { attachJsonlLineReader, serializeJsonLine } from "../rpc/jsonl.js"; @@ -66,6 +68,7 @@ import { CompactAssistantStreamReconstructor, isCompactAssistantDelta } from "./ import { DAEMON_CATALOG_ROLE_ENV, DaemonCatalogClient } from "./daemon-catalog-process.js"; import { deserializeDaemonError, serializeDaemonError } from "./daemon-errors.js"; import { + cloneCallerOwnedSessionLaunchEnv, collectDaemonClientEnv, createDaemonEventMeta, DAEMON_COMMAND_COMPATIBILITY, @@ -344,6 +347,8 @@ interface ResidentWorker { intentionalStop: boolean; stopRevision: number; launchEnv?: Record; + launchEnvMode?: "replace"; + exactEnvironmentAwaitingOwner?: boolean; transientCreateCommand?: DaemonCreateCommand; stopOperation?: Promise; stopFinalization?: { @@ -545,10 +550,21 @@ function withoutCommandId(command: DaemonCommand): DaemonCommandBody { } function withoutSupervisorCreateFields(command: DaemonCreateCommand): DaemonCreateCommand { - const { launchEnv: _launchEnv, lifecycle: _lifecycle, ...workerCommand } = command; + const { launchEnv: _launchEnv, launchEnvMode: _launchEnvMode, lifecycle: _lifecycle, ...workerCommand } = command; return workerCommand; } +function prepareCallerOwnedCreateEnvironment(command: DaemonCreateCommand): DaemonCreateCommand { + if (command.launchEnvMode === undefined) return command; + if (command.launchEnvMode !== "replace" || command.lifecycle !== "client_owned" || command.launchEnv === undefined) { + throw new Error("Exact caller-owned launch environment requires a client-owned session snapshot"); + } + return { + ...command, + launchEnv: cloneCallerOwnedSessionLaunchEnv(command.launchEnv) as Record, + }; +} + function responseWithId(response: DaemonResponse, id: string | undefined): DaemonResponse { return { ...response, id }; } @@ -577,6 +593,7 @@ function isDaemonWorkerDescriptor(value: unknown, socketPath: string): value is (descriptor.pid ?? 0) > 0 && (descriptor.processStartId === undefined || typeof descriptor.processStartId === "string") && (descriptor.ownerClientId === undefined || typeof descriptor.ownerClientId === "string") && + (descriptor.callerOwnedEnvironmentContract === undefined || descriptor.callerOwnedEnvironmentContract === true) && typeof descriptor.socketPath === "string" && typeof descriptor.authenticationToken === "string" && typeof descriptor.rootActiveSessionId === "string" && @@ -1109,6 +1126,8 @@ export class DaemonSupervisor { authorizedActiveSessionIds: new Set([durableDescriptor.rootActiveSessionId]), intentionalStop: durableDescriptor.stopRequestedAt !== undefined, stopRevision: 0, + launchEnvMode: durableDescriptor.callerOwnedEnvironmentContract === true ? "replace" : undefined, + exactEnvironmentAwaitingOwner: durableDescriptor.callerOwnedEnvironmentContract === true, }; this.workers.set(durableDescriptor.workerId, worker); try { @@ -1887,20 +1906,21 @@ export class DaemonSupervisor { case "list_saved_sessions": return this.handleSavedSessionList(client, command); case "create": { - const worker = await this.createOrReuseWorker(this.protocolClientId(client), command); + const createCommand = prepareCallerOwnedCreateEnvironment(command); + const worker = await this.createOrReuseWorker(this.protocolClientId(client), createCommand); // The owner may disconnect while creation is waiting for daemon readiness, // process launch, or session materialization. Re-evaluate ownership only // after the worker is registered so the disconnect edge cannot be missed. this.scheduleOwnedWorkerCleanup(worker); - const requestedSummary = command.sessionPath - ? this.findSummaryInWorker(worker, command.sessionPath) + const requestedSummary = createCommand.sessionPath + ? this.findSummaryInWorker(worker, createCommand.sessionPath) : undefined; if ( requestedSummary && (requestedSummary.activeSessionId ?? requestedSummary.id) !== worker.descriptor.rootActiveSessionId ) { // A create forwarded to a recovering worker still surfaces an opaque lifecycle error. - const response = await this.forwardToWorker(worker, withoutSupervisorCreateFields(command)); + const response = await this.forwardToWorker(worker, withoutSupervisorCreateFields(createCommand)); if (response.success && isSessionSummary(response.data)) { await this.refreshWorkerSummaries(worker); return { ...response, id: command.id, data: this.publicSummary(worker, response.data) }; @@ -2133,16 +2153,18 @@ export class DaemonSupervisor { case "get_owned_session_cleanup": return success(command.id, command.type, this.getOwnedSessionCleanup(command.activeSessionId)); case "complete_owned_session": { - const match = await this.findWorkerForClient(client, command.activeSessionId); + const match = await this.findWorker(command.activeSessionId); if (match.worker.descriptor.ownerClientId !== this.protocolClientId(client)) { - throw new Error("Session is not owned by this client"); + return failure(command.id, command.type, "Session is not owned by this client", { + code: "owned_session_owner_mismatch", + }); } if (match.worker.ownerCleanupTimer) { clearTimeout(match.worker.ownerCleanupTimer); match.worker.ownerCleanupTimer = undefined; } await this.stopWorker(match.worker, true); - return success(command.id, command.type); + return success(command.id, command.type, { status: "completed" as const }); } case "promote_owned_session": { const match = await this.findWorkerForClient(client, command.activeSessionId); @@ -2834,7 +2856,8 @@ export class DaemonSupervisor { throw new Error("Session is not owned by this client"); } const previousDescriptor = worker.descriptor; - worker.descriptor = { ...previousDescriptor, ownerClientId: undefined }; + const { callerOwnedEnvironmentContract: _contract, ...promotedDescriptor } = previousDescriptor; + worker.descriptor = { ...promotedDescriptor, ownerClientId: undefined }; try { this.persistWorker(worker); } catch (error) { @@ -2847,6 +2870,8 @@ export class DaemonSupervisor { worker.ownerCleanupTimer = undefined; } worker.launchEnv = undefined; + worker.launchEnvMode = undefined; + worker.exactEnvironmentAwaitingOwner = undefined; worker.transientCreateCommand = undefined; } @@ -2864,7 +2889,15 @@ export class DaemonSupervisor { } }; assertLaunchCurrent(); - const launchEnv = command.launchEnv ?? existing?.launchEnv; + const launchEnvMode = command.launchEnvMode ?? existing?.launchEnvMode; + const candidateLaunchEnv = command.launchEnv ?? existing?.launchEnv; + const launchEnv = + launchEnvMode === "replace" && candidateLaunchEnv + ? (cloneCallerOwnedSessionLaunchEnv(candidateLaunchEnv) as Record) + : candidateLaunchEnv; + if (launchEnvMode === "replace" && !launchEnv) { + throw new Error("Exact caller-owned launch environment snapshot is unavailable"); + } const createCommand: DaemonCreateCommand = { ...withoutSupervisorCreateFields(command), config: mergeAgentSessionRuntimeConfig(this.defaultSessionConfig, command.config), @@ -2880,9 +2913,9 @@ export class DaemonSupervisor { const orphanProcessJournalPath = existing?.descriptor.orphanProcessJournalPath ?? join(this.descriptorDir, `${workerId}.orphans.jsonl`); const launch = createCliSubprocessLaunchSpec(["--mode", "daemon", "--daemon-socket", socketPath]); - const workerEnvironment = createCliSubprocessEnv({ - ...process.env, - ...launchEnv, + const workerEnvironmentSource = { + ...(launchEnvMode === "replace" ? launchEnv : { ...process.env, ...launchEnv }), + // Prime-owned worker authentication, recovery, startup, lease, and orphan-cleanup bootstrap. [DAEMON_WORKER_ROLE_ENV]: "1", [DAEMON_WORKER_TOKEN_ENV]: token, [DAEMON_WORKER_ACTIVE_SESSION_ID_ENV]: rootActiveSessionId, @@ -2892,8 +2925,10 @@ export class DaemonSupervisor { [ORPHAN_PROCESS_JOURNAL_ENV]: orphanProcessJournalPath, [SESSION_LEASES_ENABLED_ENV]: "1", [SESSION_LEASE_OWNER_ID_ENV]: rootActiveSessionId, - }); - delete workerEnvironment.RLM_DEPTH; + }; + const workerEnvironment: NodeJS.ProcessEnv = + launchEnvMode === "replace" ? workerEnvironmentSource : createCliSubprocessEnv(workerEnvironmentSource); + if (launchEnvMode !== "replace") delete workerEnvironment.RLM_DEPTH; await this.assertRecoveryAllowed(); assertLaunchCurrent(); const child: ChildProcess = spawn(launch.command, launch.args, { @@ -2946,6 +2981,9 @@ export class DaemonSupervisor { authenticationToken: token, rootActiveSessionId, ownerClientId: existing?.descriptor.ownerClientId ?? ownerClientId, + ...(launchEnvMode === "replace" || existing?.descriptor.callerOwnedEnvironmentContract === true + ? { callerOwnedEnvironmentContract: true as const } + : {}), sessionDir: createCommand.config?.sessionDir, telemetryDisabled: createCommand.config?.telemetryDisabled, createdAt: existing?.descriptor.createdAt ?? now, @@ -2967,13 +3005,21 @@ export class DaemonSupervisor { intentionalStop: false, stopRevision: 0, launchEnv, - transientCreateCommand: ownerClientId ? createCommand : undefined, + launchEnvMode, + exactEnvironmentAwaitingOwner: false, + transientCreateCommand: ownerClientId + ? { ...createCommand, launchEnv, ...(launchEnvMode ? { launchEnvMode } : {}) } + : undefined, }; await this.assertRecoveryAllowed(); assertLaunchCurrent(); worker.descriptor = descriptor; worker.launchEnv = launchEnv; - worker.transientCreateCommand = descriptor.ownerClientId ? createCommand : undefined; + worker.launchEnvMode = launchEnvMode; + worker.exactEnvironmentAwaitingOwner = false; + worker.transientCreateCommand = descriptor.ownerClientId + ? { ...createCommand, launchEnv, ...(launchEnvMode ? { launchEnvMode } : {}) } + : undefined; descriptorAssigned = true; this.persistWorker(worker); worker.intentionalStop = false; @@ -3885,7 +3931,8 @@ export class DaemonSupervisor { worker: ResidentWorker, recovery: ReadonlyMap, ): Promise { - const client = this.requireAvailableWorkerClient(worker); + const client = worker.client; + if (!client) throw new Error("Session worker is not connected"); for (const [activeSessionId, recovered] of recovery) { const summary = worker.summaries.get(activeSessionId); if (!summary || summary.sessionId !== recovered.sessionId) { @@ -4472,14 +4519,71 @@ export class DaemonSupervisor { (worker.descriptor.rootActiveSessionId === command.activeSessionId || worker.descriptor.rootSessionId === command.activeSessionId), ); + if ( + ownedWorker?.descriptor.ownerClientId !== undefined && + ownedWorker.descriptor.ownerClientId !== this.protocolClientId(client) + ) { + throw new Error(`Unknown active session: ${command.activeSessionId}`); + } + const requestsExactEnvironment = + command.capabilities?.includes(CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE) === true; + if (requestsExactEnvironment !== (command.launchEnvMode === "replace")) { + throw new Error("Caller-owned environment capability and launch mode must agree"); + } + if (requestsExactEnvironment && !ownedWorker) { + throw new Error("Unknown active session"); + } + if ( + requestsExactEnvironment && + (command.launchEnv === undefined || + command.recoveryConfig === undefined || + ownedWorker?.descriptor.callerOwnedEnvironmentContract !== true) + ) { + throw new Error( + "Exact caller-owned launch environment requires a contract-created owned session and recovery context", + ); + } + const exactLaunchEnv = requestsExactEnvironment + ? (cloneCallerOwnedSessionLaunchEnv(command.launchEnv!) as Record) + : undefined; if (ownedWorker) { - if (ownedWorker.descriptor.ownerClientId !== this.protocolClientId(client)) { - throw new Error(`Unknown active session: ${command.activeSessionId}`); - } this.assertTelemetryAttachAllowed(ownedWorker, command.telemetryDisabled); - ownedWorker.launchEnv = command.launchEnv ?? ownedWorker.launchEnv; + if (exactLaunchEnv) { + if (ownedWorker.launchEnv) { + if (!isDeepStrictEqual(ownedWorker.launchEnv, exactLaunchEnv)) { + throw new Error("Caller-owned launch environment does not match the established snapshot"); + } + } else { + if (ownedWorker.exactEnvironmentAwaitingOwner !== true) { + throw new Error("Caller-owned launch environment cannot be rebound"); + } + ownedWorker.launchEnv = exactLaunchEnv; + ownedWorker.exactEnvironmentAwaitingOwner = false; + } + ownedWorker.transientCreateCommand = { + ...ownedWorker.descriptor.createCommand, + config: { + ...command.recoveryConfig!, + ...(ownedWorker.descriptor.telemetryDisabled === true ? { telemetryDisabled: true } : {}), + }, + env: command.env, + launchEnv: ownedWorker.launchEnv, + launchEnvMode: "replace", + lifecycle: "client_owned", + }; + if ( + ownedWorker.launchEnvMode !== "replace" || + ownedWorker.descriptor.callerOwnedEnvironmentContract !== true || + !ownedWorker.launchEnv || + !ownedWorker.transientCreateCommand.config + ) { + throw new Error("Caller-owned environment recovery state is incomplete"); + } + } else if (ownedWorker.launchEnvMode !== "replace") { + ownedWorker.launchEnv = command.launchEnv ?? ownedWorker.launchEnv; + } if (!ownedWorker.client || ownedWorker.descriptor.lifecycle !== "ready") { - if (command.recoveryConfig) { + if (!requestsExactEnvironment && command.recoveryConfig) { ownedWorker.transientCreateCommand = { ...ownedWorker.descriptor.createCommand, config: { @@ -4503,6 +4607,7 @@ export class DaemonSupervisor { await this.recoverWorker(ownedWorker); } } + const match = await this.findWorkerForClient(client, command.activeSessionId); this.assertTelemetryAttachAllowed(match.worker, command.telemetryDisabled); this.requireAvailableWorkerClient(match.worker); diff --git a/packages/coding-agent/src/modes/daemon/daemon-worker-protocol.ts b/packages/coding-agent/src/modes/daemon/daemon-worker-protocol.ts index 8c4fed73d9..ade1b370a6 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-worker-protocol.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-worker-protocol.ts @@ -108,6 +108,8 @@ export interface DaemonWorkerDescriptor { rootActiveSessionId: string; /** Stable protocol client that owns this worker. Omitted for resident sessions. */ ownerClientId?: string; + /** Non-secret marker that the current owned worker process was launched from an exact caller snapshot. */ + callerOwnedEnvironmentContract?: true; rootSessionId?: string; sessionFile?: string; sessionDir?: string; @@ -151,6 +153,7 @@ export function durableDaemonWorkerDescriptor(descriptor: DaemonWorkerDescriptor authenticationToken: descriptor.authenticationToken, rootActiveSessionId: descriptor.rootActiveSessionId, ...(descriptor.ownerClientId !== undefined ? { ownerClientId: descriptor.ownerClientId } : {}), + ...(descriptor.callerOwnedEnvironmentContract === true ? { callerOwnedEnvironmentContract: true as const } : {}), ...(descriptor.rootSessionId !== undefined ? { rootSessionId: descriptor.rootSessionId } : {}), ...(descriptor.sessionFile !== undefined ? { sessionFile: descriptor.sessionFile } : {}), ...(sessionDir !== undefined ? { sessionDir } : {}), diff --git a/packages/coding-agent/src/modes/daemon/snapshot-transcript-cache.ts b/packages/coding-agent/src/modes/daemon/snapshot-transcript-cache.ts index 1e24109c9c..b45b0b7278 100644 --- a/packages/coding-agent/src/modes/daemon/snapshot-transcript-cache.ts +++ b/packages/coding-agent/src/modes/daemon/snapshot-transcript-cache.ts @@ -66,10 +66,15 @@ function assertSafeSnapshotCacheParent(parentRoot: string): void { if (!metadata.isDirectory() || metadata.isSymbolicLink()) { throw new Error(`Snapshot cache parent is not a private directory: ${parentRoot}`); } - if (typeof process.getuid === "function" && metadata.uid !== process.getuid()) { + const getuid = process.getuid; + const supportsPosixOwnership = typeof getuid === "function"; + if (supportsPosixOwnership && metadata.uid !== getuid()) { throw new Error(`Snapshot cache parent is not owned by the current user: ${parentRoot}`); } - if ((metadata.mode & 0o077) !== 0) { + // Node does not expose Windows DACLs through mode bits, and chmod cannot + // make those synthesized bits private. Windows still rejects links here; + // access is governed by the containing directory's DACL, which this code does not inspect. + if (supportsPosixOwnership && (metadata.mode & 0o077) !== 0) { chmodSync(parentRoot, 0o700); const secured = lstatSync(parentRoot); if (!secured.isDirectory() || secured.isSymbolicLink() || (secured.mode & 0o077) !== 0) { @@ -118,7 +123,7 @@ export function createSnapshotCacheProcessRoot(parentRoot: string, ownerToken = } throw new Error(`Snapshot cache process root is unsafe: ${root}`); } - if ((metadata.mode & 0o077) !== 0) chmodSync(root, 0o700); + if (typeof process.getuid === "function" && (metadata.mode & 0o077) !== 0) chmodSync(root, 0o700); return root; } diff --git a/packages/coding-agent/src/modes/index.ts b/packages/coding-agent/src/modes/index.ts index 025476e1ed..d19a6bf3ea 100644 --- a/packages/coding-agent/src/modes/index.ts +++ b/packages/coding-agent/src/modes/index.ts @@ -2,7 +2,11 @@ * Run modes for the coding agent. */ -export { PRIME_AGENT_SDK_FEATURES, type PrimeAgentSdkFeature } from "../sdk-features.js"; +export { + CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE, + PRIME_AGENT_SDK_FEATURES, + type PrimeAgentSdkFeature, +} from "../sdk-features.js"; export { type AcpModeOptions, acpStopReason, @@ -29,6 +33,11 @@ export type { AgentConnectionSessionEvent, AgentConnectionSlashCommand, AgentConnectionState, + DaemonAgentConnectionOptions, + DaemonOwnedSessionContractProof, + DaemonOwnedSessionDaemonIdentity, + DaemonOwnedSessionDisposeOptions, + DaemonOwnedSessionDisposeResult, } from "./agent-connection/index.js"; export { DaemonAgentConnection, InProcessAgentConnection } from "./agent-connection/index.js"; export { type AgentsViewModeOptions, runAgentsViewMode } from "./agents-view/agents-view-mode.js"; @@ -68,6 +77,7 @@ export { DaemonClient, type DaemonClientMessageListener, type DaemonClientOptions, + type DaemonClientRequestOptions, DaemonInboundFrameTooLargeError, DEFAULT_DAEMON_CLIENT_MAX_INBOUND_FRAME_BYTES, } from "./daemon/daemon-client.js"; @@ -87,6 +97,8 @@ export type { DaemonOutbound, DaemonOwnedSessionCleanupResult, DaemonOwnedSessionCleanupStatus, + DaemonOwnedSessionCompletionResult, + DaemonOwnedSessionCompletionStatus, DaemonProtocolInfo, DaemonProtocolName, DaemonProtocolVersion, diff --git a/packages/coding-agent/src/sdk-features.ts b/packages/coding-agent/src/sdk-features.ts index f34d4eb1b4..50b73610e5 100644 --- a/packages/coding-agent/src/sdk-features.ts +++ b/packages/coding-agent/src/sdk-features.ts @@ -4,9 +4,12 @@ * These tokens describe local SDK behavior. They are not daemon capabilities, * protocol versions, schema revisions, or proof about a remote peer. */ +export const CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE = "caller_owned_session_environment_cleanup_v1" as const; + export const PRIME_AGENT_SDK_FEATURES = Object.freeze([ "bounded_daemon_ingress_v1", "negotiated_daemon_session_capabilities_v1", + CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE, ] as const); export type PrimeAgentSdkFeature = (typeof PRIME_AGENT_SDK_FEATURES)[number]; diff --git a/packages/coding-agent/test/agent-connection-daemon.test.ts b/packages/coding-agent/test/agent-connection-daemon.test.ts index fd412cea6a..85deb0bd30 100644 --- a/packages/coding-agent/test/agent-connection-daemon.test.ts +++ b/packages/coding-agent/test/agent-connection-daemon.test.ts @@ -7,6 +7,7 @@ import { SessionImportFileNotFoundError } from "../src/core/session-import-error import { DAEMON_REFINE_REQUEST_TIMEOUT_MS, DaemonAgentConnection, + type DaemonOwnedSessionContractProof, } from "../src/modes/agent-connection/daemon-agent-connection.js"; import type { AgentConnectionEvent, @@ -37,6 +38,7 @@ import { class FakeDaemonClient { readonly requests: DaemonCommand[] = []; readonly requestTimeouts: number[] = []; + readonly requestOptions: DaemonClientRequestOptions[] = []; attachResultFactory: ((command: Extract) => DaemonAttachResult) | undefined; reattachResultFactory: ((command: Extract) => DaemonAttachResult) | undefined; switchSessionActiveSessionId: string | undefined; @@ -53,6 +55,7 @@ class FakeDaemonClient { resetTransportCount = 0; reconnectError: Error | undefined; attachFailures = 0; + attachError: Error | undefined; connectionStateGate: Promise | undefined; connectionStateFactory: ((activeSessionId: string) => AgentConnectionState) | undefined; rlmChildren: AgentConnectionRlmChildAgentSnapshot[] = []; @@ -75,6 +78,12 @@ class FakeDaemonClient { cancelCorrelatedPromptResponse: unknown; cancelPromptAdmissionStatus: "cancelled" | "owned" | "unknown" = "owned"; serverCapabilities = new Set(); + ownedSessionCleanupStatus: "active" | "stopping" | "settled" = "active"; + ownedSessionCompletionStatus: "completed" | "owner_mismatch" | "invalid" = "completed"; + ownedSessionCleanupTransportFailure = false; + ownedSessionCleanupGate: Promise | undefined; + ownedSessionCompletionGate: Promise | undefined; + sideQuestionAbortGate: Promise | undefined; updateRestartSessions: Array> = []; hello: DaemonHello | undefined = { type: "daemon_hello", @@ -94,6 +103,7 @@ class FakeDaemonClient { ): Promise { this.requests.push(command); this.requestTimeouts.push(timeoutMs); + this.requestOptions.push(options); switch (command.type) { case "prompt": if (this.promptGate) await this.promptGate; @@ -166,6 +176,7 @@ class FakeDaemonClient { }; case "attach": if (this.attachGate) await this.attachGate; + if (this.attachError) throw this.attachError; if (this.attachFailures > 0) { this.attachFailures--; throw new Error("attach failed"); @@ -500,6 +511,9 @@ class FakeDaemonClient { return { type: "response", command: command.type, success: true }; case "start_side_question": return { type: "response", command: command.type, success: true }; + case "abort_side_question": + await this.sideQuestionAbortGate; + return { type: "response", command: command.type, success: true, data: { aborted: true } }; case "delete_saved_session": return { type: "response", @@ -583,6 +597,35 @@ class FakeDaemonClient { filePath: "/tmp/not-found.jsonl", }, }; + case "get_owned_session_cleanup": + await this.ownedSessionCleanupGate; + if (this.ownedSessionCleanupTransportFailure) { + this.connected = false; + throw new Error("private cleanup transport canary"); + } + return { + type: "response", + command: command.type, + success: true, + data: { status: this.ownedSessionCleanupStatus }, + }; + case "complete_owned_session": + await this.ownedSessionCompletionGate; + if (this.ownedSessionCompletionStatus === "owner_mismatch") { + return { + type: "response", + command: command.type, + success: false, + error: "Session is not owned by this client", + errorInfo: { code: "owned_session_owner_mismatch" }, + }; + } + return { + type: "response", + command: command.type, + success: true, + data: this.ownedSessionCompletionStatus === "invalid" ? { status: "invalid" } : { status: "completed" }, + }; default: throw new Error(`Unexpected command: ${command.type}`); } @@ -628,6 +671,10 @@ class FakeDaemonClient { return this.transportGeneration; } + get isConnected(): boolean { + return this.connected; + } + get isClosed(): boolean { return this.ownerClosed; } @@ -699,6 +746,23 @@ function asDaemonClient(client: FakeDaemonClient): DaemonClient { return client as unknown as DaemonClient; } +const SUPERVISOR_GENERATION_A = "11111111-1111-4111-8111-111111111111"; +const SUPERVISOR_GENERATION_B = "22222222-2222-4222-8222-222222222222"; + +function enableCallerOwnedSessionContract( + client: FakeDaemonClient, + supervisorGeneration = SUPERVISOR_GENERATION_A, +): void { + client.serverCapabilities.add("caller_owned_session_environment_cleanup_v1"); + client.serverCapabilities.add("authoritative_owned_session_cleanup_v1"); + client.hello = { + ...client.hello!, + schemaRevision: 29, + appVersion: "0.8.1-test", + supervisorGeneration, + }; +} + function createConnectionState(activeSessionId: string, sessionId: string): AgentConnectionState { return { activeSessionId, @@ -819,7 +883,9 @@ function createAttachResult( capability === "extension_ui" || capability === "slim_attach" || capability === "chunked_snapshot" || - capability === "correlated_prompt_lifecycle_v1", + capability === "correlated_prompt_lifecycle_v1" || + capability === "client_owned_sessions" || + capability === "caller_owned_session_environment_cleanup_v1", ), }, }; @@ -991,6 +1057,263 @@ describe("DaemonAgentConnection", () => { else expect(request).not.toHaveProperty("recoveryConfig"); }); + it("clones one caller-owned environment snapshot and publishes only a secret-free attach proof", async () => { + const fakeClient = new FakeDaemonClient(); + enableCallerOwnedSessionContract(fakeClient); + const source = { PROVIDER_TOKEN: "private-env-a", PATH: "/caller/a" }; + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-owned", { + ownedSession: true, + ownedSessionLaunchEnv: source, + ownedSessionRecoveryConfig: {}, + }); + source.PROVIDER_TOKEN = "private-env-c"; + source.PATH = "/ambient/c"; + + await connection.attach(); + + const attach = fakeClient.requests[0]; + expect(attach).toMatchObject({ + type: "attach", + launchEnv: { PROVIDER_TOKEN: "private-env-a", PATH: "/caller/a" }, + launchEnvMode: "replace", + capabilities: expect.arrayContaining(["caller_owned_session_environment_cleanup_v1"]), + }); + const proof = connection.getOwnedSessionContractProof(); + expect(proof).toMatchObject({ + feature: "caller_owned_session_environment_cleanup_v1", + status: "attached", + daemon: { + protocolName: "prime-agent.daemon", + protocolVersion: 7, + schemaRevision: 29, + appVersion: "0.8.1-test", + supervisorGeneration: SUPERVISOR_GENERATION_A, + transportGeneration: 1, + }, + }); + const serializedProof = JSON.stringify(proof); + expect(serializedProof).not.toContain("private-env"); + expect(serializedProof).not.toContain("/caller"); + expect(serializedProof).not.toContain("fake.sock"); + }); + + it("omits unshaped daemon identity strings and rejects an unshaped generation", async () => { + const identityCanary = "private-daemon-identity-canary/path"; + const fakeClient = new FakeDaemonClient(); + enableCallerOwnedSessionContract(fakeClient); + fakeClient.hello = { + ...fakeClient.hello!, + appVersion: identityCanary, + runtime: { buildId: identityCanary, executablePath: identityCanary }, + }; + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-owned", { + ownedSession: true, + ownedSessionLaunchEnv: { PROVIDER_TOKEN: "private" }, + ownedSessionRecoveryConfig: {}, + }); + await connection.attach(); + const proof = connection.getOwnedSessionContractProof(); + expect(proof).toBeDefined(); + expect(proof?.daemon).not.toHaveProperty("appVersion"); + expect(proof?.daemon).not.toHaveProperty("buildId"); + expect(JSON.stringify(proof)).not.toContain(identityCanary); + + const unsafeGenerationClient = new FakeDaemonClient(); + enableCallerOwnedSessionContract(unsafeGenerationClient, identityCanary); + const unsafeGenerationConnection = new DaemonAgentConnection( + asDaemonClient(unsafeGenerationClient), + "active-owned", + { + ownedSession: true, + ownedSessionLaunchEnv: { PROVIDER_TOKEN: "private" }, + ownedSessionRecoveryConfig: {}, + }, + ); + await unsafeGenerationConnection.attach(); + expect(unsafeGenerationConnection.getOwnedSessionContractProof()).toBeUndefined(); + + const oldProtocolClient = new FakeDaemonClient(); + enableCallerOwnedSessionContract(oldProtocolClient); + oldProtocolClient.hello = { + ...oldProtocolClient.hello!, + protocol: { ...oldProtocolClient.hello!.protocol, version: 6 }, + }; + const oldProtocolConnection = new DaemonAgentConnection(asDaemonClient(oldProtocolClient), "active-owned", { + ownedSession: true, + ownedSessionLaunchEnv: { PROVIDER_TOKEN: "private" }, + ownedSessionRecoveryConfig: {}, + }); + await oldProtocolConnection.attach(); + expect(oldProtocolConnection.getOwnedSessionContractProof()).toBeUndefined(); + }); + + it("allows only the fixed Pylon artifact build diagnostic shape", async () => { + const attachWithBuildId = async (buildId: string, enableContract = true) => { + const fakeClient = new FakeDaemonClient(); + if (enableContract) enableCallerOwnedSessionContract(fakeClient); + fakeClient.hello = { ...fakeClient.hello!, runtime: { buildId, executablePath: "/not-exposed" } }; + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-owned", { + ownedSession: true, + ownedSessionLaunchEnv: { OWNER_ENV: "private" }, + ownedSessionRecoveryConfig: {}, + }); + await connection.attach(); + return connection.getOwnedSessionContractProof(); + }; + + await expect(attachWithBuildId("pylon-build-g0123456789ab-r17")).resolves.toMatchObject({ + daemon: { buildId: "pylon-build-g0123456789ab-r17" }, + }); + for (const unsafe of [ + "pylon-build-g0123456789AB-r17", + "pylon-build-g0123456789ab-r0", + "pylon-build-g0123456789ab-r17/private", + ]) { + const proof = await attachWithBuildId(unsafe); + expect(proof?.daemon).not.toHaveProperty("buildId"); + } + expect(await attachWithBuildId("pylon-build-g0123456789ab-r17", false)).toBeUndefined(); + }); + + it("preserves the exact snapshot across attach retry and transport recovery", async () => { + const fakeClient = new FakeDaemonClient(); + enableCallerOwnedSessionContract(fakeClient); + fakeClient.attachFailures = 1; + const source = { PROVIDER_TOKEN: "private-env-a", PATH: "/caller/a" }; + const recoverDaemon = vi.fn(async () => undefined); + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-owned", { + ownedSession: true, + ownedSessionLaunchEnv: source, + ownedSessionRecoveryConfig: {}, + recoverDaemon, + reconnectTimeoutMs: 2000, + }); + source.PROVIDER_TOKEN = "private-env-c"; + + await connection.attach(); + fakeClient.connected = false; + fakeClient.emitClose(new Error("private transport canary")); + await vi.waitFor(() => + expect(fakeClient.requests.filter((request) => request.type === "attach")).toHaveLength(3), + ); + await vi.waitFor(() => expect(connection.getOwnedSessionContractProof()).toBeDefined()); + + for (const request of fakeClient.requests.filter((command) => command.type === "attach")) { + expect(request.launchEnv).toEqual({ PROVIDER_TOKEN: "private-env-a", PATH: "/caller/a" }); + expect(request.launchEnvMode).toBe("replace"); + expect(JSON.stringify(request)).not.toContain("private-env-c"); + } + expect(recoverDaemon).toHaveBeenCalledOnce(); + expect(connection.getOwnedSessionContractProof()?.daemon.transportGeneration).toBe(2); + }); + + it("preserves the exact snapshot and recovery context on reattach direct fallback", async () => { + const fakeClient = new FakeDaemonClient(); + enableCallerOwnedSessionContract(fakeClient); + fakeClient.serverCapabilities.add("owned_session_recovery_context"); + fakeClient.reattachResultFactory = (command) => { + const full = createAttachResult(command.targetActiveSessionId, command.clientId, command.capabilities, 1); + const snapshotId = "owned-reattach-failure"; + const { messages: _messages, ...snapshot } = full.snapshot; + queueMicrotask(() => { + fakeClient.emitMessage({ + type: "session_snapshot_begin", + activeSessionId: command.targetActiveSessionId, + snapshotId, + snapshot, + messageCount: 1, + targetChunkBytes: 512 * 1024, + purpose: "replacement", + }); + fakeClient.emitMessage({ + type: "session_snapshot_failed", + activeSessionId: command.targetActiveSessionId, + snapshotId, + error: "forced transfer failure", + purpose: "replacement", + }); + }); + return { + ...full, + snapshot: { ...full.snapshot, messages: [] }, + snapshotStream: { id: snapshotId, messageCount: 1, targetChunkBytes: 512 * 1024 }, + }; + }; + const recoveryConfig = { cwd: "/caller/project", agentDir: "/caller/home" }; + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-source", { + ownedSession: true, + ownedSessionLaunchEnv: { PROVIDER_TOKEN: "private-env-a", PATH: "/caller/a" }, + ownedSessionRecoveryConfig: recoveryConfig, + }); + await connection.attach(); + + await ( + connection as unknown as { + reattachSession(source: string, target: string): Promise<{ cancelled: false }>; + } + ).reattachSession("active-source", "active-target"); + + const recoveryRequests = fakeClient.requests.filter( + (request) => + request.type === "reattach" || (request.type === "attach" && request.activeSessionId === "active-target"), + ); + expect(recoveryRequests.map((request) => request.type)).toEqual(["reattach", "attach"]); + for (const request of recoveryRequests) { + expect(request).toMatchObject({ + launchEnv: { PROVIDER_TOKEN: "private-env-a", PATH: "/caller/a" }, + launchEnvMode: "replace", + recoveryConfig, + capabilities: expect.arrayContaining(["caller_owned_session_environment_cleanup_v1"]), + }); + } + }); + + it("rejects an exact environment without recovery context before advertising the contract", () => { + const fakeClient = new FakeDaemonClient(); + enableCallerOwnedSessionContract(fakeClient); + expect( + () => + new DaemonAgentConnection(asDaemonClient(fakeClient), "active-owned", { + ownedSession: true, + ownedSessionLaunchEnv: { PROVIDER_TOKEN: "private" }, + }), + ).toThrow("ownedSessionLaunchEnv requires ownedSessionRecoveryConfig"); + expect(fakeClient.requests).toEqual([]); + }); + + it("does not advertise or prove the contract when the server offer is absent", async () => { + const fakeClient = new FakeDaemonClient(); + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-owned", { + ownedSession: true, + ownedSessionLaunchEnv: { PROVIDER_TOKEN: "private" }, + ownedSessionRecoveryConfig: {}, + }); + + await connection.attach(); + + const attach = fakeClient.requests[0]; + if (attach?.type !== "attach") throw new Error("Missing attach request"); + expect(attach.capabilities).not.toContain("caller_owned_session_environment_cleanup_v1"); + expect(connection.getOwnedSessionContractProof()).toBeUndefined(); + }); + + it("fails closed when the contract token is offered without authoritative cleanup", async () => { + const fakeClient = new FakeDaemonClient(); + fakeClient.serverCapabilities.add("caller_owned_session_environment_cleanup_v1"); + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-owned", { + ownedSession: true, + ownedSessionLaunchEnv: { PROVIDER_TOKEN: "private" }, + ownedSessionRecoveryConfig: {}, + }); + + await connection.attach(); + + const attach = fakeClient.requests[0]; + if (attach?.type !== "attach") throw new Error("Missing attach request"); + expect(attach.capabilities).not.toContain("caller_owned_session_environment_cleanup_v1"); + expect(connection.getOwnedSessionContractProof()).toBeUndefined(); + }); + it("forwards queueIfBusy for prompt admission", async () => { const fakeClient = new FakeDaemonClient(); const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-1"); @@ -1778,6 +2101,339 @@ describe("DaemonAgentConnection", () => { expect(connection.supportsNegotiatedCapability("correlated_prompt_lifecycle_v1")).toBe(false); }); + it("returns fixed observable owned-session cleanup outcomes", async () => { + const createConnection = async ( + cleanup: FakeDaemonClient["ownedSessionCleanupStatus"], + completion: FakeDaemonClient["ownedSessionCompletionStatus"] = "completed", + ) => { + const fakeClient = new FakeDaemonClient(); + enableCallerOwnedSessionContract(fakeClient); + fakeClient.ownedSessionCleanupStatus = cleanup; + fakeClient.ownedSessionCompletionStatus = completion; + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-owned", { + ownedSession: true, + ownedSessionLaunchEnv: { PROVIDER_TOKEN: "private-cleanup-canary" }, + ownedSessionRecoveryConfig: {}, + }); + await connection.attach(); + return { connection, fakeClient }; + }; + + const completed = await createConnection("active"); + await expect(completed.connection.disposeOwnedSession({ timeoutMs: 100 })).resolves.toMatchObject({ + status: "completed", + daemonReplaced: false, + }); + + const already = await createConnection("settled"); + await expect(already.connection.disposeOwnedSession({ timeoutMs: 100 })).resolves.toMatchObject({ + status: "already_completed", + daemonReplaced: false, + }); + expect(already.fakeClient.requests.filter((request) => request.type === "complete_owned_session")).toEqual([]); + + const mismatch = await createConnection("active", "owner_mismatch"); + await expect(mismatch.connection.disposeOwnedSession({ timeoutMs: 100 })).resolves.toMatchObject({ + status: "owner_mismatch", + }); + + const active = await createConnection("active", "invalid"); + await expect(active.connection.disposeOwnedSession({ timeoutMs: 5 })).resolves.toMatchObject({ + status: "uncertain", + reason: "active", + }); + + const stopping = await createConnection("stopping"); + await expect(stopping.connection.disposeOwnedSession({ timeoutMs: 5 })).resolves.toMatchObject({ + status: "uncertain", + reason: "stopping", + }); + + const failed = await createConnection("active"); + failed.fakeClient.ownedSessionCleanupTransportFailure = true; + const failedResult = await failed.connection.disposeOwnedSession({ timeoutMs: 100 }); + expect(failedResult).toMatchObject({ status: "transport_failure" }); + expect(JSON.stringify(failedResult)).not.toContain("private cleanup transport canary"); + expect(JSON.stringify(failedResult)).not.toContain("private-cleanup-canary"); + + const unsupportedClient = new FakeDaemonClient(); + const unsupported = new DaemonAgentConnection(asDaemonClient(unsupportedClient), "active-owned", { + ownedSession: true, + }); + await unsupported.attach(); + await expect(unsupported.disposeOwnedSession({ timeoutMs: 100 })).resolves.toEqual({ + feature: "caller_owned_session_environment_cleanup_v1", + status: "unsupported", + }); + }); + + it("uses one total owned cleanup deadline for hung side-question aborts", async () => { + const fakeClient = new FakeDaemonClient(); + enableCallerOwnedSessionContract(fakeClient); + fakeClient.ownedSessionCleanupStatus = "settled"; + fakeClient.sideQuestionAbortGate = new Promise(() => {}); + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-owned", { + ownedSession: true, + ownedSessionLaunchEnv: { OWNER_ENV: "exact" }, + ownedSessionRecoveryConfig: {}, + }); + await connection.attach(); + await connection.startSideQuestion("side-hung", "question", []); + + const startedAt = Date.now(); + await expect(connection.disposeOwnedSession({ timeoutMs: 20 })).resolves.toMatchObject({ + status: "already_completed", + }); + expect(Date.now() - startedAt).toBeLessThan(250); + const abortIndex = fakeClient.requests.findIndex((request) => request.type === "abort_side_question"); + expect(abortIndex).toBeGreaterThanOrEqual(0); + expect(fakeClient.requestTimeouts[abortIndex]).toBeGreaterThan(0); + expect(fakeClient.requestTimeouts[abortIndex]).toBeLessThanOrEqual(20); + expect(fakeClient.requestOptions[abortIndex]?.recoverAcrossReconnect).toBe(false); + }); + + it("bounds unsupported peer completion by the same total cleanup deadline", async () => { + const fakeClient = new FakeDaemonClient(); + fakeClient.ownedSessionCompletionGate = new Promise(() => {}); + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-legacy", { + ownedSession: true, + }); + await connection.attach(); + + const startedAt = Date.now(); + await expect(connection.disposeOwnedSession({ timeoutMs: 20 })).resolves.toMatchObject({ status: "unsupported" }); + expect(Date.now() - startedAt).toBeLessThan(250); + const completionIndex = fakeClient.requests.findIndex((request) => request.type === "complete_owned_session"); + expect(completionIndex).toBeGreaterThanOrEqual(0); + expect(fakeClient.requestTimeouts[completionIndex]).toBeGreaterThan(0); + expect(fakeClient.requestTimeouts[completionIndex]).toBeLessThanOrEqual(20); + expect(fakeClient.requestOptions[completionIndex]?.recoverAcrossReconnect).toBe(false); + }); + + it("does not reuse an owned cleanup proof for a different pending route", async () => { + const fakeClient = new FakeDaemonClient(); + enableCallerOwnedSessionContract(fakeClient); + fakeClient.ownedSessionCleanupStatus = "settled"; + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-a", { + ownedSession: true, + ownedSessionLaunchEnv: { PROVIDER_TOKEN: "private" }, + ownedSessionRecoveryConfig: {}, + }); + await connection.attach(); + (connection as unknown as { pendingReattachActiveSessionIds: Set }).pendingReattachActiveSessionIds.add( + "active-b", + ); + + await expect(connection.disposeOwnedSession({ timeoutMs: 100 })).resolves.toMatchObject({ + status: "unsupported", + }); + expect(fakeClient.requests.filter((request) => request.type === "get_owned_session_cleanup")).toEqual([]); + expect(fakeClient.requests.filter((request) => request.type === "complete_owned_session")).toEqual([]); + }); + + it("joins concurrent observable cleanup calls into one completion", async () => { + const fakeClient = new FakeDaemonClient(); + enableCallerOwnedSessionContract(fakeClient); + let releaseCompletion!: () => void; + fakeClient.ownedSessionCompletionGate = new Promise((resolve) => { + releaseCompletion = resolve; + }); + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-owned", { + ownedSession: true, + ownedSessionLaunchEnv: { PROVIDER_TOKEN: "private" }, + ownedSessionRecoveryConfig: {}, + }); + await connection.attach(); + + const first = connection.disposeOwnedSession({ timeoutMs: 1000 }); + const second = connection.disposeOwnedSession({ timeoutMs: 1000 }); + await vi.waitFor(() => + expect(fakeClient.requests.filter((request) => request.type === "complete_owned_session")).toHaveLength(1), + ); + releaseCompletion(); + + const [firstResult, secondResult] = await Promise.all([first, second]); + expect(firstResult).toEqual(secondResult); + expect(firstResult.status).toBe("completed"); + expect(fakeClient.requests.filter((request) => request.type === "complete_owned_session")).toHaveLength(1); + }); + + it("invalidates proof on disconnect and proves replacement settlement only after reattach", async () => { + const fakeClient = new FakeDaemonClient(); + enableCallerOwnedSessionContract(fakeClient, SUPERVISOR_GENERATION_A); + fakeClient.ownedSessionCleanupStatus = "settled"; + let releaseCleanup!: () => void; + fakeClient.ownedSessionCleanupGate = new Promise((resolve) => { + releaseCleanup = resolve; + }); + let releaseRecovery!: () => void; + const recoveryGate = new Promise((resolve) => { + releaseRecovery = resolve; + }); + const recoverDaemon = vi.fn(async () => { + await recoveryGate; + fakeClient.hello = { ...fakeClient.hello!, supervisorGeneration: SUPERVISOR_GENERATION_B }; + }); + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-owned", { + ownedSession: true, + ownedSessionLaunchEnv: { PROVIDER_TOKEN: "private" }, + ownedSessionRecoveryConfig: {}, + recoverDaemon, + reconnectTimeoutMs: 2000, + }); + await connection.attach(); + expect(connection.getOwnedSessionContractProof()).toBeDefined(); + fakeClient.connected = false; + fakeClient.emitClose(new Error("replacement")); + expect(connection.getOwnedSessionContractProof()).toBeUndefined(); + await vi.waitFor(() => expect(recoverDaemon).toHaveBeenCalledOnce()); + + const disposing = connection.disposeOwnedSession({ timeoutMs: 1000 }); + releaseRecovery(); + await vi.waitFor(() => + expect(fakeClient.requests.filter((request) => request.type === "attach")).toHaveLength(2), + ); + expect(connection.getOwnedSessionContractProof()).toBeUndefined(); + expect( + (connection as unknown as { currentOwnedSessionContractProof?: unknown }).currentOwnedSessionContractProof, + ).toBeUndefined(); + expect( + (connection as unknown as { lastOwnedSessionContractProof?: DaemonOwnedSessionContractProof }) + .lastOwnedSessionContractProof?.daemon.supervisorGeneration, + ).toBe(SUPERVISOR_GENERATION_A); + releaseCleanup(); + await expect(disposing).resolves.toMatchObject({ + status: "replacement_settled", + daemonReplaced: true, + started: { daemon: { supervisorGeneration: SUPERVISOR_GENERATION_A } }, + observed: { supervisorGeneration: SUPERVISOR_GENERATION_B }, + }); + expect(connection.getOwnedSessionContractProof()).toBeUndefined(); + }); + + it("fences an in-flight completion to its proved transport and queries replacement state fresh", async () => { + const fakeClient = new FakeDaemonClient(); + enableCallerOwnedSessionContract(fakeClient, SUPERVISOR_GENERATION_A); + fakeClient.ownedSessionCleanupStatus = "active"; + let releaseCompletion!: () => void; + fakeClient.ownedSessionCompletionGate = new Promise((resolve) => { + releaseCompletion = resolve; + }); + const recoverDaemon = vi.fn(async () => { + fakeClient.hello = { ...fakeClient.hello!, supervisorGeneration: SUPERVISOR_GENERATION_B }; + }); + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-owned", { + ownedSession: true, + ownedSessionLaunchEnv: { OWNER_ENV: "private" }, + ownedSessionRecoveryConfig: {}, + recoverDaemon, + reconnectTimeoutMs: 2000, + }); + await connection.attach(); + + const disposing = connection.disposeOwnedSession({ timeoutMs: 1000 }); + await vi.waitFor(() => + expect(fakeClient.requests.filter((request) => request.type === "complete_owned_session")).toHaveLength(1), + ); + fakeClient.connected = false; + fakeClient.emitClose(new Error("completion transport closed")); + fakeClient.ownedSessionCleanupStatus = "settled"; + await vi.waitFor(() => + expect(fakeClient.requests.filter((request) => request.type === "attach")).toHaveLength(2), + ); + expect(connection.getOwnedSessionContractProof()).toBeUndefined(); + expect(fakeClient.requests.filter((request) => request.type === "get_owned_session_cleanup")).toHaveLength(1); + releaseCompletion(); + + await expect(disposing).resolves.toMatchObject({ + status: "replacement_settled", + daemonReplaced: true, + observed: { supervisorGeneration: SUPERVISOR_GENERATION_B }, + }); + expect(fakeClient.requests.filter((request) => request.type === "complete_owned_session")).toHaveLength(1); + expect(fakeClient.requests.filter((request) => request.type === "get_owned_session_cleanup")).toHaveLength(2); + for (let index = 1; index < fakeClient.requests.length; index++) { + const request = fakeClient.requests[index]!; + if ( + request.type === "attach" || + request.type === "get_owned_session_cleanup" || + request.type === "complete_owned_session" + ) { + expect(fakeClient.requestOptions[index]?.recoverAcrossReconnect).toBe(false); + } + } + }); + + it("reports same-supervisor reconnect settlement without treating transport churn as replacement", async () => { + const fakeClient = new FakeDaemonClient(); + enableCallerOwnedSessionContract(fakeClient, SUPERVISOR_GENERATION_A); + fakeClient.ownedSessionCleanupStatus = "settled"; + let releaseRecovery!: () => void; + const recoveryGate = new Promise((resolve) => { + releaseRecovery = resolve; + }); + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-owned", { + ownedSession: true, + ownedSessionLaunchEnv: { OWNER_ENV: "private" }, + ownedSessionRecoveryConfig: {}, + recoverDaemon: async () => recoveryGate, + reconnectTimeoutMs: 2000, + }); + await connection.attach(); + fakeClient.connected = false; + fakeClient.emitClose(new Error("same supervisor reconnect")); + await vi.waitFor(() => expect(connection.getOwnedSessionContractProof()).toBeUndefined()); + + const disposing = connection.disposeOwnedSession({ timeoutMs: 1000 }); + releaseRecovery(); + await expect(disposing).resolves.toMatchObject({ + status: "already_completed", + daemonReplaced: false, + started: { daemon: { supervisorGeneration: SUPERVISOR_GENERATION_A, transportGeneration: 1 } }, + observed: { supervisorGeneration: SUPERVISOR_GENERATION_A, transportGeneration: 2 }, + }); + expect(connection.getOwnedSessionContractProof()).toBeUndefined(); + expect(fakeClient.requests.filter((request) => request.type === "complete_owned_session")).toEqual([]); + }); + + it("uses a replacement transport for read-only settlement when the owned worker is gone", async () => { + const fakeClient = new FakeDaemonClient(); + enableCallerOwnedSessionContract(fakeClient, SUPERVISOR_GENERATION_A); + fakeClient.ownedSessionCleanupStatus = "settled"; + let releaseRecovery!: () => void; + const recoveryGate = new Promise((resolve) => { + releaseRecovery = resolve; + }); + const recoverDaemon = vi.fn(async () => { + await recoveryGate; + fakeClient.hello = { ...fakeClient.hello!, supervisorGeneration: SUPERVISOR_GENERATION_B }; + fakeClient.attachError = new Error("Unknown active session"); + }); + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-owned", { + ownedSession: true, + ownedSessionLaunchEnv: { OWNER_ENV: "private" }, + ownedSessionRecoveryConfig: {}, + recoverDaemon, + reconnectTimeoutMs: 2000, + }); + await connection.attach(); + fakeClient.connected = false; + fakeClient.emitClose(new Error("replacement without worker")); + await vi.waitFor(() => expect(recoverDaemon).toHaveBeenCalledOnce()); + + const disposing = connection.disposeOwnedSession({ timeoutMs: 1000 }); + releaseRecovery(); + await expect(disposing).resolves.toMatchObject({ + status: "replacement_settled", + daemonReplaced: true, + observed: { supervisorGeneration: SUPERVISOR_GENERATION_B }, + }); + expect(connection.getOwnedSessionContractProof()).toBeUndefined(); + expect(fakeClient.requests.filter((request) => request.type === "get_owned_session_cleanup")).toHaveLength(1); + expect(fakeClient.requests.filter((request) => request.type === "complete_owned_session")).toEqual([]); + expect(fakeClient.resetTransportCount).toBe(0); + }); + it("rejects attach re-entry throughout the owned-session dispose wait", async () => { const fakeClient = new FakeDaemonClient(); fakeClient.serverCapabilities.add("correlated_prompt_lifecycle_v1"); diff --git a/packages/coding-agent/test/daemon-client.test.ts b/packages/coding-agent/test/daemon-client.test.ts index ac7c09a407..31c38a9b46 100644 --- a/packages/coding-agent/test/daemon-client.test.ts +++ b/packages/coding-agent/test/daemon-client.test.ts @@ -1,6 +1,9 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { + DaemonAgentConnectionOptions as RootDaemonAgentConnectionOptions, DaemonClientOptions as RootDaemonClientOptions, + DaemonClientRequestOptions as RootDaemonClientRequestOptions, + DaemonOwnedSessionDisposeResult as RootDaemonOwnedSessionDisposeResult, PrimeAgentSdkFeature as RootPrimeAgentSdkFeature, } from "../src/index.js"; import * as publicSdk from "../src/index.js"; @@ -209,10 +212,28 @@ describe("DaemonClient", () => { const rootFeatures: RootPrimeAgentSdkFeature[] = [ "bounded_daemon_ingress_v1", "negotiated_daemon_session_capabilities_v1", + "caller_owned_session_environment_cleanup_v1", ]; + const rootConnectionOptions: RootDaemonAgentConnectionOptions = { + ownedSession: true, + ownedSessionLaunchEnv: { PROVIDER_TOKEN: "private" }, + ownedSessionRecoveryConfig: {}, + }; + const rootRequestOptions: RootDaemonClientRequestOptions = { recoverAcrossReconnect: false }; + const rootDisposeResult: RootDaemonOwnedSessionDisposeResult = { + feature: "caller_owned_session_environment_cleanup_v1", + status: "unsupported", + }; expect(Array.isArray(publicSdk.PRIME_AGENT_SDK_FEATURES)).toBe(true); expect(publicSdk.PRIME_AGENT_SDK_FEATURES).toEqual(rootFeatures); expect(Object.isFrozen(publicSdk.PRIME_AGENT_SDK_FEATURES)).toBe(true); + expect(publicSdk.CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE).toBe( + "caller_owned_session_environment_cleanup_v1", + ); + expect(rootConnectionOptions.ownedSession).toBe(true); + expect(rootConnectionOptions.ownedSessionRecoveryConfig).toEqual({}); + expect(rootRequestOptions.recoverAcrossReconnect).toBe(false); + expect(rootDisposeResult.status).toBe("unsupported"); expect(publicSdk.DaemonClient).toBe(DaemonClient); expect(publicSdk.DaemonInboundFrameTooLargeError).toBe(DaemonInboundFrameTooLargeError); expect(publicSdk.DEFAULT_DAEMON_CLIENT_MAX_INBOUND_FRAME_BYTES).toBe( @@ -630,6 +651,46 @@ describe("DaemonClient", () => { client.close(); }); + it("gates exact caller-owned launch replacement while preserving legacy launch commands", async () => { + const oldClient = new DaemonClient("/tmp/prime-agent-old.sock"); + const oldConnect = oldClient.connect(); + const oldSocket = netMock.sockets.at(-1)!; + oldSocket.emit("connect"); + await oldConnect; + emitHello(oldSocket, DAEMON_PROTOCOL_VERSION, ["client_owned_sessions"], 28); + + await expect( + oldClient.request({ + type: "create", + lifecycle: "client_owned", + launchEnv: { PROVIDER_TOKEN: "private" }, + launchEnvMode: "replace", + }), + ).rejects.toThrow("does not support caller_owned_session_environment_cleanup_v1"); + expect(oldSocket.writes).toEqual([]); + oldClient.close(); + + const legacyClient = new DaemonClient("/tmp/prime-agent-new.sock"); + const legacyConnect = legacyClient.connect(); + const legacySocket = netMock.sockets.at(-1)!; + legacySocket.emit("connect"); + await legacyConnect; + emitHello( + legacySocket, + DAEMON_PROTOCOL_VERSION, + ["client_owned_sessions", "caller_owned_session_environment_cleanup_v1"], + DAEMON_SCHEMA_REVISION, + ); + const request = legacyClient.request({ + type: "create", + lifecycle: "client_owned", + launchEnv: { PROVIDER_TOKEN: "legacy" }, + }); + await vi.waitFor(() => expect(legacySocket.writes).toHaveLength(1)); + legacyClient.close(); + await expect(request).rejects.toThrow("closed before the operation completed"); + }); + it("field-gates prompt admissionId before writing raw commands", async () => { const client = new DaemonClient("/tmp/prime-agent.sock"); const connect = client.connect(); @@ -1145,6 +1206,37 @@ describe("DaemonClient", () => { client.close(); }); + it("rejects only requests that opt out of reconnect recovery while preserving legacy replay", async () => { + const client = new DaemonClient("/tmp/prime-agent.sock"); + client.enableRequestRecovery(); + const firstConnect = client.connect(); + const firstSocket = netMock.sockets[0]!; + firstSocket.emit("connect"); + await firstConnect; + emitHello(firstSocket); + + const ordinary = client.request({ type: "list" }); + const ordinaryWireData = firstSocket.writes[0]!; + const ordinaryEnvelope = JSON.parse(ordinaryWireData) as { id: string }; + const transportBound = client.request({ type: "list" }, 30_000, { recoverAcrossReconnect: false }); + expect(firstSocket.writes).toHaveLength(2); + firstSocket.emit("close"); + + await expect(transportBound).rejects.toThrow("Connection to the Prime Agent daemon closed"); + const secondConnect = client.connect(); + const secondSocket = netMock.sockets[1]!; + secondSocket.emit("connect"); + await secondConnect; + emitHello(secondSocket); + expect(secondSocket.writes).toEqual([ordinaryWireData]); + secondSocket.emit( + "data", + `${JSON.stringify({ id: ordinaryEnvelope.id, type: "response", command: "list", success: true })}\n`, + ); + await expect(ordinary).resolves.toMatchObject({ id: ordinaryEnvelope.id, success: true }); + client.close(); + }); + it("pauses request timeouts while a recoverable connection is disconnected", async () => { vi.useFakeTimers(); const client = new DaemonClient("/tmp/prime-agent.sock"); diff --git a/packages/coding-agent/test/daemon-protocol.test.ts b/packages/coding-agent/test/daemon-protocol.test.ts index 526f00eeab..ae3ac362e4 100644 --- a/packages/coding-agent/test/daemon-protocol.test.ts +++ b/packages/coding-agent/test/daemon-protocol.test.ts @@ -3,6 +3,8 @@ import { readFileSync } from "node:fs"; import { resolve } from "node:path"; import { describe, expect, it } from "vitest"; import { + cloneCallerOwnedSessionLaunchEnv, + collectDaemonLaunchEnv, createDaemonCommandEnvelope, createDaemonEventEnvelope, createDaemonEventMeta, @@ -30,6 +32,7 @@ import { type DaemonWorkerDescriptor, durableDaemonWorkerDescriptor, } from "../src/modes/daemon/daemon-worker-protocol.js"; +import { CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE } from "../src/sdk-features.js"; describe("daemon protocol helpers", () => { it("serializes worker descriptors as identity-only version 2 state", () => { @@ -44,6 +47,7 @@ describe("daemon protocol helpers", () => { supervisorSocketPath: "/tmp/supervisor.sock", authenticationToken: "local-worker-token", rootActiveSessionId: "active", + callerOwnedEnvironmentContract: true, sessionFile: "/sessions/root.jsonl", createdAt: "2026-01-01T00:00:00.000Z", updatedAt: "2026-01-01T00:00:00.000Z", @@ -72,6 +76,7 @@ describe("daemon protocol helpers", () => { expect(durable.createCommand).toEqual({ type: "create", sessionPath: "/sessions/root.jsonl" }); expect(durable).toMatchObject({ workerId: "worker", + callerOwnedEnvironmentContract: true, sessionFile: "/sessions/root.jsonl", sessionDir: "/legacy/sessions", telemetryDisabled: true, @@ -79,6 +84,103 @@ describe("daemon protocol helpers", () => { expect(JSON.stringify(durable)).not.toContain("secret-"); }); + it("defensively clones exact caller-owned environments without exposing rejected values", () => { + const source = { PROVIDER_TOKEN: "secret-a", Path: "/caller/bin" }; + const snapshot = cloneCallerOwnedSessionLaunchEnv(source, "linux"); + source.PROVIDER_TOKEN = "secret-c"; + + expect(snapshot).toEqual({ PROVIDER_TOKEN: "secret-a", Path: "/caller/bin" }); + expect(Object.isFrozen(snapshot)).toBe(true); + expect(cloneCallerOwnedSessionLaunchEnv({ Path: "a", PATH: "b" }, "linux")).toEqual({ + Path: "a", + PATH: "b", + }); + expect(() => cloneCallerOwnedSessionLaunchEnv({ Path: "a", PATH: "b" }, "win32")).toThrow( + "duplicate Windows environment keys", + ); + for (const invalid of [ + { "": "value" }, + { "BAD=NAME": "value" }, + { "BAD\0NAME": "value" }, + { BAD_VALUE: "value\0suffix" }, + { prime_agent_internal_private: "value" }, + { BAD_TYPE: 1 } as unknown as Record, + ]) { + expect(() => cloneCallerOwnedSessionLaunchEnv(invalid)).toThrow(TypeError); + } + const accessor = {} as Record; + Object.defineProperty(accessor, "PRIVATE", { get: () => "value" }); + expect(() => cloneCallerOwnedSessionLaunchEnv(accessor)).toThrow(TypeError); + const symbolKeyed = {} as Record; + Object.defineProperty(symbolKeyed, Symbol("private"), { value: "value" }); + expect(() => cloneCallerOwnedSessionLaunchEnv(symbolKeyed)).toThrow("string environment keys"); + const inherited = Object.create({ INHERITED_PRIVATE: "ignored" }) as Record; + expect(cloneCallerOwnedSessionLaunchEnv(inherited)).toEqual({}); + const reserved = "private-environment-canary"; + const error = (() => { + try { + cloneCallerOwnedSessionLaunchEnv({ PRIME_AGENT_INTERNAL_PRIVATE: reserved }); + } catch (cause) { + return cause; + } + })(); + expect(error).toBeInstanceOf(TypeError); + expect(String(error)).not.toContain(reserved); + + const proxyCanary = "hostile-proxy-private-canary"; + const hostileProxy = new Proxy(Object.create(null) as Record, { + ownKeys() { + throw new Error(proxyCanary); + }, + }); + let proxyError: unknown; + try { + cloneCallerOwnedSessionLaunchEnv(hostileProxy); + } catch (cause) { + proxyError = cause; + } + expect(proxyError).toBeInstanceOf(TypeError); + expect(String(proxyError)).toBe( + "TypeError: ownedSessionLaunchEnv must be an inspectable string environment record", + ); + expect(String(proxyError)).not.toContain(proxyCanary); + const revoked = Proxy.revocable(Object.create(null) as Record, {}); + revoked.revoke(); + expect(() => cloneCallerOwnedSessionLaunchEnv(revoked.proxy)).toThrow( + "ownedSessionLaunchEnv must be an inspectable string environment record", + ); + + for (const key of ["RLM_DEPTH", "rlm_depth", "Rlm_Depth"]) { + expect(() => cloneCallerOwnedSessionLaunchEnv({ [key]: "private-depth" })).toThrow(TypeError); + } + expect( + collectDaemonLaunchEnv({ RLM_DEPTH: "1", rlm_depth: "2", KEEP_EXACT: "yes" } as NodeJS.ProcessEnv), + ).toEqual({ KEEP_EXACT: "yes" }); + }); + + it("capability-gates exact launch replacement without changing legacy commands", () => { + const exact = getDaemonCommandCompatibilities({ + type: "create", + lifecycle: "client_owned", + launchEnv: { PROVIDER_TOKEN: "private" }, + launchEnvMode: "replace", + }); + expect(exact).toEqual([ + { + minProtocol: 7, + minSchemaRevision: 29, + capability: CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE, + }, + { minProtocol: 7 }, + ]); + expect(getDaemonCommandCompatibilities({ type: "create", launchEnv: { PROVIDER_TOKEN: "legacy" } })).toEqual([ + { minProtocol: 7 }, + ]); + expect(DAEMON_SUPPORTED_CLIENT_CAPABILITIES).toContain(CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE); + expect(DAEMON_DEFAULT_SERVER_CAPABILITIES).not.toContain(CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE); + expect(DAEMON_SUPERVISOR_SERVER_CAPABILITIES).toContain(CALLER_OWNED_SESSION_ENVIRONMENT_CLEANUP_FEATURE); + }); + it("keeps the advertised schema identity synchronized with wire type shapes", () => { const source = readFileSync(resolve(__dirname, "../src/modes/daemon/daemon-protocol.ts"), "utf8"); const commandSource = source.slice( @@ -93,8 +195,22 @@ describe("daemon protocol helpers", () => { source.indexOf("export type DaemonOutbound ="), source.indexOf("export const DAEMON_OUTBOUND_COMPATIBILITY"), ); + const ownedSessionSource = source.slice( + source.indexOf("export type DaemonOwnedSessionCleanupStatus"), + source.indexOf("export type DaemonServerCapability"), + ); + const launchEnvironmentSource = source.slice( + source.indexOf("export type DaemonLaunchEnvMode"), + source.indexOf("/**\n * The allowlist of env vars"), + ); + const errorInfoSource = source.slice( + source.indexOf("export type DaemonErrorInfo ="), + source.indexOf("export type DaemonSessionClosedReason"), + ); const digest = createHash("sha256") - .update(`${commandSource}\n${savedSessionSource}\n${outboundSource}`) + .update( + `${commandSource}\n${savedSessionSource}\n${outboundSource}\n${ownedSessionSource}\n${launchEnvironmentSource}\n${errorInfoSource}`, + ) .digest("hex") .slice(0, 12); expect(DAEMON_SCHEMA_ID).toBe(`protocol-${DAEMON_PROTOCOL_VERSION}-schema-${DAEMON_SCHEMA_REVISION}-${digest}`); @@ -173,7 +289,7 @@ describe("daemon protocol helpers", () => { }); it("capability- and schema-gates fresh snapshot generation nonces", () => { - expect(DAEMON_SCHEMA_REVISION).toBe(28); + expect(DAEMON_SCHEMA_REVISION).toBe(29); expect(DAEMON_SNAPSHOT_GENERATION_NONCE_MIN_SCHEMA_REVISION).toBe(28); expect( getDaemonCommandCompatibilities({ diff --git a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts index e2a8225785..3cacd67258 100644 --- a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts @@ -38,7 +38,7 @@ const workerLaunchTestState = vi.hoisted(() => ({ gateMarkerPath: "", tsxCliPath: "", cliEntrypoint: "", - spawned: [] as Array<{ child: ChildProcess; args: readonly string[] }>, + spawned: [] as Array<{ child: ChildProcess; args: readonly string[]; options: SpawnOptions }>, })); vi.mock("node:child_process", async (importOriginal) => { @@ -50,7 +50,7 @@ vi.mock("node:child_process", async (importOriginal) => { spawn(command: string, args: readonly string[], options: SpawnOptions): ChildProcess { const child = actual.spawn(command, args, options); if (workerLaunchTestState.capture) { - workerLaunchTestState.spawned.push({ child, args }); + workerLaunchTestState.spawned.push({ child, args, options }); } return child; }, @@ -516,6 +516,82 @@ describe("daemon worker supervisor monitoring", () => { expect(handleWorkerCommand).not.toHaveBeenCalled(); }); + it("replaces ambient supervisor env only for capability-marked caller-owned launches", async () => { + workerLaunchTestState.capture = true; + workerLaunchTestState.fixtureMode = "rollback-gate"; + const root = mkdtempSync(join(tmpdir(), "prime-supervisor-exact-env-test-")); + const descriptorDir = join(root, "descriptors"); + mkdirSync(descriptorDir, { recursive: true }); + supervisorRegistryDirs.add(root); + const launchOnce = async (exact: boolean) => { + let assertionCount = 0; + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + ...createSupervisorSnapshotState(), + defaultSessionConfig: { cwd: root, agentDir: root }, + descriptorDir, + socketPath: join(root, `${exact ? "exact" : "legacy"}.sock`), + workers: new Map(), + assertRecoveryAllowed: vi.fn(async () => { + assertionCount++; + if (assertionCount === 3) throw new Error("stop after env capture"); + }), + log: vi.fn(), + }) as { + launchWorker(command: { + type: "create"; + config: { cwd: string; agentDir: string }; + lifecycle: "client_owned"; + launchEnv: Record; + launchEnvMode?: "replace"; + }): Promise; + }; + await expect( + supervisor.launchWorker({ + type: "create", + config: { cwd: root, agentDir: root }, + lifecycle: "client_owned", + launchEnv: { CALLER_ENV_A: "a" }, + ...(exact ? { launchEnvMode: "replace" as const } : {}), + }), + ).rejects.toThrow("stop after env capture"); + }; + const previousAmbient = process.env.SUPERVISOR_ENV_C; + const previousExecArgv = process.execArgv; + process.env.SUPERVISOR_ENV_C = "c"; + process.execArgv = [...previousExecArgv, "--import=tsx"]; + try { + await launchOnce(true); + await launchOnce(false); + } finally { + process.execArgv = previousExecArgv; + if (previousAmbient === undefined) delete process.env.SUPERVISOR_ENV_C; + else process.env.SUPERVISOR_ENV_C = previousAmbient; + } + + const exactEnvironment = workerLaunchTestState.spawned[0]?.options.env; + const legacyEnvironment = workerLaunchTestState.spawned[1]?.options.env; + expect(exactEnvironment).toMatchObject({ CALLER_ENV_A: "a", PRIME_AGENT_INTERNAL_DAEMON_WORKER: "1" }); + expect(Object.keys(exactEnvironment ?? {}).sort()).toEqual( + [ + "CALLER_ENV_A", + "PRIME_AGENT_INTERNAL_DAEMON_SUPERVISOR_SOCKET", + "PRIME_AGENT_INTERNAL_DAEMON_WORKER", + "PRIME_AGENT_INTERNAL_DAEMON_WORKER_ACTIVE_SESSION_ID", + "PRIME_AGENT_INTERNAL_DAEMON_WORKER_RECOVERY_JOURNAL", + "PRIME_AGENT_INTERNAL_DAEMON_WORKER_STARTUP_GATE_FD", + "PRIME_AGENT_INTERNAL_DAEMON_WORKER_TOKEN", + "PRIME_AGENT_INTERNAL_ORPHAN_PROCESS_JOURNAL", + "PRIME_AGENT_INTERNAL_SESSION_LEASES", + "PRIME_AGENT_INTERNAL_SESSION_LEASE_OWNER_ID", + ].sort(), + ); + expect(exactEnvironment).not.toHaveProperty("SUPERVISOR_ENV_C"); + expect(exactEnvironment).not.toHaveProperty("TSX_TSCONFIG_PATH"); + expect(legacyEnvironment).toMatchObject({ CALLER_ENV_A: "a", SUPERVISOR_ENV_C: "c" }); + expect(legacyEnvironment).toHaveProperty("TSX_TSCONFIG_PATH"); + expect(JSON.stringify(exactEnvironment)).not.toContain("private"); + }); + it.each([ { name: "first post-spawn ownership check", @@ -2469,6 +2545,65 @@ describe("daemon worker supervisor monitoring", () => { if (worker.ownerCleanupTimer) clearTimeout(worker.ownerCleanupTimer); }); + it("returns structured completion ownership without claiming arbitrary settlement", async () => { + const owner = { id: "socket-owner" } as DaemonSocketClient; + const stranger = { id: "socket-stranger" } as DaemonSocketClient; + const worker = { + descriptor: { + workerId: "owned-worker", + ownerClientId: "protocol-owner", + rootActiveSessionId: "active-owned", + }, + summaries: new Map([ + [ + "active-owned", + { + id: "session-owned", + sessionId: "session-owned", + activeSessionId: "active-owned", + sessionName: "owned-friendly", + }, + ], + ]), + ownerCleanupTimer: undefined, + }; + const stopWorker = vi.fn(async () => undefined); + const protocolClientIds = new WeakMap([ + [owner, "protocol-owner"], + [stranger, "protocol-stranger"], + ]); + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + workers: new Map([[worker.descriptor.workerId, worker]]), + protocolClientIds, + stopWorker, + }) as { + handleCommand( + client: DaemonSocketClient, + command: { type: "complete_owned_session"; activeSessionId: string }, + ): Promise<{ success: boolean; data?: unknown }>; + }; + + const wrongOwner = await supervisor.handleCommand(stranger, { + type: "complete_owned_session", + activeSessionId: "active-owned", + }); + expect(wrongOwner).toMatchObject({ + success: false, + error: "Session is not owned by this client", + errorInfo: { code: "owned_session_owner_mismatch" }, + }); + expect(wrongOwner).not.toHaveProperty("data"); + expect(JSON.stringify(wrongOwner)).not.toContain("active-owned"); + expect(stopWorker).not.toHaveBeenCalled(); + await expect( + supervisor.handleCommand(owner, { type: "complete_owned_session", activeSessionId: "owned-friendly" }), + ).resolves.toMatchObject({ success: true, data: { status: "completed" } }); + expect(stopWorker).toHaveBeenCalledWith(worker, true); + await expect( + supervisor.handleCommand(owner, { type: "complete_owned_session", activeSessionId: "never-owned" }), + ).rejects.toThrow("Unknown active session: never-owned"); + }); + it("joins an admitted recovery before proving exact cleanup settled", async () => { const recoveryGate = createDeferred(); let staleRecoveryRejected = false; @@ -3928,6 +4063,7 @@ describe("daemon worker supervisor monitoring", () => { descriptor: { workerId: "worker-owned-recovery", ownerClientId: "client-1", + callerOwnedEnvironmentContract: true, rootActiveSessionId: activeSessionId, lifecycle: "failed", consecutiveFailures: 1, @@ -3938,11 +4074,13 @@ describe("daemon worker supervisor monitoring", () => { intentionalStop: false, stopRevision: 0, launchEnv: undefined as Record | undefined, + launchEnvMode: "replace" as "replace" | undefined, + exactEnvironmentAwaitingOwner: true, transientCreateCommand: undefined as Record | undefined, }; const client = { id: "client-1", - capabilities: new Set(), + capabilities: new Set(["caller_owned_session_environment_cleanup_v1"]), supportsExtensionUi: false, attachedActiveSessionIds: new Set(), }; @@ -3963,6 +4101,8 @@ describe("daemon worker supervisor monitoring", () => { activeSessionId: string; recoveryConfig: { cwd: string }; launchEnv: Record; + launchEnvMode: "replace"; + capabilities: readonly ["caller_owned_session_environment_cleanup_v1"]; env: Record; }, ): Promise; @@ -3974,6 +4114,8 @@ describe("daemon worker supervisor monitoring", () => { activeSessionId, recoveryConfig: { cwd: "/tmp/fresh-owner" }, launchEnv: { OWNER_SECRET: "fresh" }, + launchEnvMode: "replace", + capabilities: ["caller_owned_session_environment_cleanup_v1"], env: { HERDR_PANE_ID: "pane-1" }, }), ).rejects.toThrow("stop after reconstruction"); @@ -3983,12 +4125,142 @@ describe("daemon worker supervisor monitoring", () => { config: { cwd: "/tmp/fresh-owner", telemetryDisabled: true }, env: { HERDR_PANE_ID: "pane-1" }, launchEnv: { OWNER_SECRET: "fresh" }, + launchEnvMode: "replace", lifecycle: "client_owned", }); expect(worker.launchEnv).toEqual({ OWNER_SECRET: "fresh" }); + expect(worker.launchEnvMode).toBe("replace"); expect(recoverWorker).toHaveBeenCalledWith(worker); }); + it("stages exact recovery for an already-ready adopted worker and rejects environment rebinding", async () => { + const activeSessionId = "active-owned-adopted"; + const summary = { id: "session-adopted", sessionId: "session-adopted", activeSessionId }; + const worker = { + descriptor: { + workerId: "worker-owned-adopted", + ownerClientId: "client-adopted", + callerOwnedEnvironmentContract: true, + rootActiveSessionId: activeSessionId, + lifecycle: "ready", + consecutiveFailures: 0, + createCommand: { type: "create" as const, sessionPath: "/tmp/adopted.jsonl" }, + }, + client: {}, + summaries: new Map([[activeSessionId, summary]]), + launchEnv: undefined as Record | undefined, + launchEnvMode: "replace" as const, + exactEnvironmentAwaitingOwner: true, + transientCreateCommand: undefined as Record | undefined, + }; + const client = { + id: "client-adopted", + capabilities: new Set(), + supportsExtensionUi: false, + attachedActiveSessionIds: new Set(), + }; + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + workers: new Map([[worker.descriptor.workerId, worker]]), + clients: new Set([client]), + protocolClientIds: new WeakMap(), + findWorkerForClient: vi.fn(async () => ({ worker, summary })), + requireAvailableWorkerClient: vi.fn(() => { + throw new Error("stop after ready staging"); + }), + }) as { + attachClient(attachClient: typeof client, command: Record): Promise; + }; + const attach = (launchEnv: Record) => + supervisor.attachClient(client, { + type: "attach", + activeSessionId, + recoveryConfig: { cwd: "/tmp/adopted-owner" }, + launchEnv, + launchEnvMode: "replace", + capabilities: ["caller_owned_session_environment_cleanup_v1"], + }); + + const stranger = { ...client, id: "client-stranger", attachedActiveSessionIds: new Set() }; + await expect( + supervisor.attachClient(stranger, { + type: "attach", + activeSessionId, + capabilities: ["caller_owned_session_environment_cleanup_v1"], + }), + ).rejects.toThrow(`Unknown active session: ${activeSessionId}`); + + await expect( + supervisor.attachClient(client, { + type: "attach", + activeSessionId, + launchEnv: { OWNER_SECRET: "original" }, + launchEnvMode: "replace", + capabilities: ["caller_owned_session_environment_cleanup_v1"], + }), + ).rejects.toThrow("requires a contract-created owned session and recovery context"); + expect(worker.launchEnv).toBeUndefined(); + expect(worker.exactEnvironmentAwaitingOwner).toBe(true); + + await expect(attach({ OWNER_SECRET: "original" })).rejects.toThrow("stop after ready staging"); + expect(worker.launchEnv).toEqual({ OWNER_SECRET: "original" }); + expect(worker.exactEnvironmentAwaitingOwner).toBe(false); + expect(worker.transientCreateCommand).toMatchObject({ + config: { cwd: "/tmp/adopted-owner" }, + launchEnv: { OWNER_SECRET: "original" }, + launchEnvMode: "replace", + lifecycle: "client_owned", + }); + await expect(attach({ OWNER_SECRET: "replacement" })).rejects.toThrow( + "Caller-owned launch environment does not match the established snapshot", + ); + expect(worker.launchEnv).toEqual({ OWNER_SECRET: "original" }); + }); + + it("refuses exact proof for a legacy-owned worker", async () => { + const activeSessionId = "active-legacy-owned"; + const worker = { + descriptor: { + workerId: "worker-legacy-owned", + ownerClientId: "client-1", + rootActiveSessionId: activeSessionId, + lifecycle: "ready", + consecutiveFailures: 0, + createCommand: { type: "create" as const }, + }, + }; + const client = { + id: "client-1", + capabilities: new Set(["caller_owned_session_environment_cleanup_v1"]), + supportsExtensionUi: false, + attachedActiveSessionIds: new Set(), + }; + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + workers: new Map([[worker.descriptor.workerId, worker]]), + protocolClientIds: new WeakMap(), + }) as { + attachClient( + attachClient: typeof client, + command: { + type: "attach"; + activeSessionId: string; + launchEnv: Record; + launchEnvMode: "replace"; + capabilities: readonly ["caller_owned_session_environment_cleanup_v1"]; + }, + ): Promise; + }; + + await expect( + supervisor.attachClient(client, { + type: "attach", + activeSessionId, + launchEnv: { OWNER_ONLY: "snapshot" }, + launchEnvMode: "replace", + capabilities: ["caller_owned_session_environment_cleanup_v1"], + }), + ).rejects.toThrow("contract-created owned session"); + }); + it("rejects an opted-out attach to a telemetry-enabled worker", async () => { const activeSessionId = "active-telemetry-enabled"; const summary = { diff --git a/packages/coding-agent/test/daemon-supervisor-process.test.ts b/packages/coding-agent/test/daemon-supervisor-process.test.ts index c919d82d59..12d1bd1d37 100644 --- a/packages/coding-agent/test/daemon-supervisor-process.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-process.test.ts @@ -10,13 +10,14 @@ import { AgentCronJobStore } from "../src/core/cron-jobs.js"; import { readActiveOrphanProcesses } from "../src/core/orphan-process-journal.js"; import { acquireSessionLease, + getProcessStartId, SESSION_LEASE_OWNER_ID_ENV, SESSION_LEASES_ENABLED_ENV, } from "../src/core/session-lease.js"; import { readSessionInfo, SessionManager } from "../src/core/session-manager.js"; import { DaemonAgentConnection } from "../src/modes/agent-connection/daemon-agent-connection.js"; import { DaemonClient, getDaemonSocketCloseReason } from "../src/modes/daemon/daemon-client.js"; -import { createDaemonCommandEnvelope } from "../src/modes/daemon/daemon-protocol.js"; +import { collectDaemonLaunchEnv, createDaemonCommandEnvelope } from "../src/modes/daemon/daemon-protocol.js"; import type { SessionSummary } from "../src/modes/daemon/daemon-session-list.js"; import type { DaemonWorkerDescriptor } from "../src/modes/daemon/daemon-worker-protocol.js"; @@ -26,6 +27,7 @@ const blockingProcessPath = resolve(__dirname, "fixtures/blocking-process.mjs"); const tempDirs: string[] = []; const children = new Set(); const workerPids = new Set(); +const identityTrackedProcesses = new Map(); const daemonSockets = new Set(); const childDiagnostics = new WeakMap(); const PROCESS_STRESS_WORKERS = Number.parseInt(process.env.PRIME_AGENT_STRESS_WORKERS ?? "10", 10); @@ -49,6 +51,26 @@ afterEach(async () => { } } children.clear(); + for (const [pid, processStartId] of identityTrackedProcesses) { + const identity = { pid, processStartId }; + try { + signalIdentityVerifiedProcess(identity, "SIGCONT"); + signalIdentityVerifiedProcess(identity, "SIGTERM"); + } catch { + continue; + } + try { + await waitForProcessGone(pid, 1000); + } catch { + try { + signalIdentityVerifiedProcess(identity, "SIGKILL"); + await waitForProcessGone(pid, 1000); + } catch { + // Gone or no longer the process identity tracked by this test. + } + } + } + identityTrackedProcesses.clear(); for (const pid of workerPids) { try { process.kill(pid, "SIGCONT"); @@ -71,6 +93,19 @@ afterEach(async () => { } }); +function signalIdentityVerifiedProcess( + identity: { pid?: number; processStartId?: string }, + signal: NodeJS.Signals, +): void { + const { pid, processStartId } = identity; + if (!pid || !processStartId) throw new Error("Process identity is incomplete"); + const observedProcessStartId = getProcessStartId(pid); + if (!observedProcessStartId || observedProcessStartId !== processStartId) { + throw new Error("Process identity changed before the requested signal"); + } + process.kill(pid, signal); +} + function tempDir(): string { const directory = mkdtempSync(join(tmpdir(), "prime-daemon-supervisor-test-")); tempDirs.push(directory); @@ -140,6 +175,20 @@ function readWorkerDescriptor(agentDir: string): DaemonWorkerDescriptor { throw new Error("Worker descriptor was not persisted"); } +function readWorkerDescriptors(agentDir: string): DaemonWorkerDescriptor[] { + const workersRoot = join(agentDir, "daemon-workers"); + try { + return readdirSync(workersRoot).flatMap((directory) => { + const descriptorDirectory = join(workersRoot, directory); + return readdirSync(descriptorDirectory) + .filter((name) => name.endsWith(".json")) + .map((name) => JSON.parse(readFileSync(join(descriptorDirectory, name), "utf8")) as DaemonWorkerDescriptor); + }); + } catch { + return []; + } +} + function countWorkerDescriptors(agentDir: string): number { const workersRoot = join(agentDir, "daemon-workers"); try { @@ -451,8 +500,8 @@ async function waitForExit(child: ChildProcess): Promise { }); } -async function waitForProcessGone(pid: number): Promise { - const deadline = Date.now() + 10_000; +async function waitForProcessGone(pid: number, timeoutMs = 10_000): Promise { + const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { try { process.kill(pid, 0); @@ -463,7 +512,7 @@ async function waitForProcessGone(pid: number): Promise { } await new Promise((resolveDelay) => setTimeout(resolveDelay, 25)); } - throw new Error(`Worker ${pid} remained alive after daemon shutdown`); + throw new Error(`Process ${pid} remained alive after daemon shutdown`); } async function waitForCondition(predicate: () => boolean, failureMessage: string, timeoutMs = 10_000): Promise { @@ -522,7 +571,15 @@ describe("daemon supervisor resident workers", () => { const client = await connectEventually(socketPath, supervisor); const created = await client.request({ type: "create", - config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, + config: { + cwd: projectDir, + agentDir, + sessionDir, + provider: "faux", + model: "faux", + noTools: true, + noExtensions: true, + }, }); if (!created.success) throw new Error(created.error); const summary = requireSummary(created.data); @@ -1019,6 +1076,414 @@ describe("daemon supervisor resident workers", () => { await waitForSocketGone(socketPath); }, 60_000); + it.skipIf(process.platform === "win32")( + "isolates two caller-owned environments through worker and supervisor recovery", + async () => { + const root = tempDir(); + const agentDir = join(root, "agent"); + const projectDir = join(root, "project"); + const socketPath = join(tmpdir(), `prime-issue33-${process.pid}-${randomUUID().slice(0, 8)}.sock`); + const observerModule = join(root, "observe-launch-env.cjs"); + mkdirSync(projectDir, { recursive: true }); + writeFileSync( + observerModule, + `const fs = require("node:fs"); +const path = process.env.PRIME_AGENT_TEST_ENV_OBSERVATION; +if (path) fs.appendFileSync(path, JSON.stringify({ pid: process.pid, role: process.env.PRIME_AGENT_INTERNAL_DAEMON_WORKER, hasA: process.env.PRIME_AGENT_TEST_ENV_A !== undefined, hasB: process.env.PRIME_AGENT_TEST_ENV_B !== undefined, hasC: process.env.PRIME_AGENT_TEST_SUPERVISOR_C !== undefined }) + "\\n"); +`, + ); + const sessionDir = join(agentDir, "sessions"); + const sessionFiles = { + A: createSnapshotSessionFile(agentDir, projectDir, "issue #33 owner A"), + B: createSnapshotSessionFile(agentDir, projectDir, "issue #33 owner B"), + }; + const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); + let replacementSupervisor: ChildProcess | undefined; + let daemonRecovery: Promise | undefined; + const recoverDaemon = () => { + daemonRecovery ??= (async () => { + replacementSupervisor = spawnSupervisor(agentDir, socketPath, projectDir, [], { + PRIME_AGENT_TEST_SUPERVISOR_C: "ambient-c", + }); + const probe = await connectEventually(socketPath, replacementSupervisor); + probe.close(); + })(); + return daemonRecovery; + }; + const safeBaseEnvironment = collectDaemonLaunchEnv(process.env); + for (const name of Object.keys(safeBaseEnvironment)) { + if (name.startsWith("RLM_") || /TOKEN|KEY|SECRET|PASSWORD|CREDENTIAL|AUTH|COOKIE/i.test(name)) { + delete safeBaseEnvironment[name]; + } + } + Object.assign(safeBaseEnvironment, { + HOME: process.env.HOME ?? root, + PI_OFFLINE: "1", + TSX_TSCONFIG_PATH: resolve(__dirname, "../../../tsconfig.json"), + [ENV_AGENT_DIR]: agentDir, + }); + delete safeBaseEnvironment.PRIME_AGENT_TEST_SUPERVISOR_C; + delete safeBaseEnvironment.PRIME_AGENT_TEST_ENV_A; + delete safeBaseEnvironment.PRIME_AGENT_TEST_ENV_B; + const createOwned = async (label: "A" | "B") => { + const client = await connectEventually(socketPath, supervisor); + const observationPath = join(root, `observed-${label}.jsonl`); + const canary = `issue33-private-${label.toLowerCase()}-${randomUUID()}`; + const launchEnv = { + ...safeBaseEnvironment, + NODE_OPTIONS: `${process.env.NODE_OPTIONS ?? ""} --require=${observerModule}`.trim(), + PRIME_AGENT_TEST_ENV_OBSERVATION: observationPath, + ...(label === "A" ? { PRIME_AGENT_TEST_ENV_A: canary } : { PRIME_AGENT_TEST_ENV_B: canary }), + }; + const created = await client.request({ + type: "create", + lifecycle: "client_owned", + sessionPath: sessionFiles[label], + launchEnv, + launchEnvMode: "replace", + config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, + }); + if (!created.success) throw new Error(created.error); + const summary = requireSummary(created.data); + if (!summary.workerPid) throw new Error("Owned worker did not expose its pid"); + const connection = await DaemonAgentConnection.attach(client, summary.activeSessionId ?? summary.id, { + ownedSession: true, + ownedSessionLaunchEnv: launchEnv, + ownedSessionRecoveryConfig: { + cwd: projectDir, + agentDir, + sessionDir, + provider: "faux", + model: "faux", + noTools: true, + noExtensions: true, + }, + recoverDaemon, + reconnectTimeoutMs: 30_000, + supportsExtensionUi: false, + }); + const proof = connection.getOwnedSessionContractProof(); + if (!proof) throw new Error("Owned attach did not publish the issue #33 contract proof"); + return { client, connection, summary, launchEnv, observationPath, canary, proof }; + }; + const ownerA = await createOwned("A"); + const ownerB = await createOwned("B"); + const readObservations = (path: string) => { + try { + return readFileSync(path, "utf8") + .trim() + .split("\n") + .filter(Boolean) + .map( + (line) => + JSON.parse(line) as { + pid: number; + role?: string; + hasA: boolean; + hasB: boolean; + hasC: boolean; + }, + ); + } catch { + return []; + } + }; + await waitForCondition( + () => readObservations(ownerA.observationPath).some((entry) => entry.role === "1"), + "Worker A did not report its launch environment", + ); + await waitForCondition( + () => readObservations(ownerB.observationPath).some((entry) => entry.role === "1"), + "Worker B did not report its launch environment", + ); + for (const entry of readObservations(ownerA.observationPath).filter( + (observation) => observation.role === "1", + )) { + expect(entry).toMatchObject({ hasA: true, hasB: false, hasC: false }); + } + for (const entry of readObservations(ownerB.observationPath).filter( + (observation) => observation.role === "1", + )) { + expect(entry).toMatchObject({ hasA: false, hasB: true, hasC: false }); + } + + const activeA = ownerA.summary.activeSessionId ?? ownerA.summary.id; + const activeB = ownerB.summary.activeSessionId ?? ownerB.summary.id; + const originalDescriptors = readWorkerDescriptors(agentDir); + const originalA = originalDescriptors.find((descriptor) => descriptor.rootActiveSessionId === activeA); + const originalB = originalDescriptors.find((descriptor) => descriptor.rootActiveSessionId === activeB); + if (!originalA?.pid || !originalA.processStartId || !originalB?.pid || !originalB.processStartId) { + throw new Error("Owned worker descriptors did not contain complete process identities"); + } + identityTrackedProcesses.set(originalA.pid, originalA.processStartId); + identityTrackedProcesses.set(originalB.pid, originalB.processStartId); + + const helloA = ownerA.client.hello; + const helloB = ownerB.client.hello; + if ( + !helloA?.supervisorPid || + !helloA.supervisorProcessStartId || + helloB?.supervisorPid !== helloA.supervisorPid || + helloB.supervisorProcessStartId !== helloA.supervisorProcessStartId + ) { + throw new Error("Owned clients did not agree on the authenticated supervisor process identity"); + } + signalIdentityVerifiedProcess( + { pid: helloA.supervisorPid, processStartId: helloA.supervisorProcessStartId }, + "SIGKILL", + ); + await waitForProcessGone(helloA.supervisorPid); + await waitForExit(supervisor); + children.delete(supervisor); + await waitForCondition( + () => { + const nextA = ownerA.connection.getOwnedSessionContractProof(); + const nextB = ownerB.connection.getOwnedSessionContractProof(); + return ( + nextA !== undefined && + nextB !== undefined && + nextA.daemon.supervisorGeneration !== ownerA.proof.daemon.supervisorGeneration && + nextB.daemon.supervisorGeneration !== ownerB.proof.daemon.supervisorGeneration + ); + }, + "Owned connections did not republish proof after supervisor replacement", + 30_000, + ); + if (!replacementSupervisor) throw new Error("Shared daemon recovery did not spawn a replacement supervisor"); + const replacementHelloA = ownerA.client.hello; + const replacementHelloB = ownerB.client.hello; + if ( + !replacementHelloA?.supervisorPid || + !replacementHelloA.supervisorProcessStartId || + replacementHelloB?.supervisorPid !== replacementHelloA.supervisorPid || + replacementHelloB.supervisorProcessStartId !== replacementHelloA.supervisorProcessStartId || + getProcessStartId(replacementHelloA.supervisorPid) !== replacementHelloA.supervisorProcessStartId + ) { + throw new Error("Replacement supervisor identity was not current and shared"); + } + identityTrackedProcesses.set(replacementHelloA.supervisorPid, replacementHelloA.supervisorProcessStartId); + + const previousAmbientC = process.env.PRIME_AGENT_TEST_SUPERVISOR_C; + process.env.PRIME_AGENT_TEST_SUPERVISOR_C = "mutated-c"; + let recoveredA: DaemonWorkerDescriptor | undefined; + try { + signalIdentityVerifiedProcess(originalA, "SIGKILL"); + await waitForProcessGone(originalA.pid); + identityTrackedProcesses.delete(originalA.pid); + await waitForCondition( + () => + readWorkerDescriptors(agentDir).some( + (descriptor) => + descriptor.rootActiveSessionId === activeA && + descriptor.pid !== originalA.pid && + descriptor.lifecycle === "ready", + ), + "Worker A was not recovered with a new process after supervisor replacement", + 20_000, + ); + recoveredA = readWorkerDescriptors(agentDir).find( + (descriptor) => + descriptor.rootActiveSessionId === activeA && + descriptor.pid !== originalA.pid && + descriptor.lifecycle === "ready", + ); + if (!recoveredA?.pid || !recoveredA.processStartId) { + throw new Error("Recovered worker A descriptor did not contain a complete process identity"); + } + identityTrackedProcesses.set(recoveredA.pid, recoveredA.processStartId); + await waitForCondition( + () => + readObservations(ownerA.observationPath).filter( + (entry) => entry.role === "1" && entry.hasA && !entry.hasB && !entry.hasC, + ).length >= 2, + "Recovered worker A did not retain its exact environment", + 20_000, + ); + } finally { + if (previousAmbientC === undefined) delete process.env.PRIME_AGENT_TEST_SUPERVISOR_C; + else process.env.PRIME_AGENT_TEST_SUPERVISOR_C = previousAmbientC; + } + if (!recoveredA?.pid) throw new Error("Recovered worker A was unavailable after recovery"); + expect(ownerA.connection.getOwnedSessionContractProof()?.daemon.supervisorGeneration).toBe( + replacementHelloA.supervisorGeneration, + ); + expect(ownerB.connection.getOwnedSessionContractProof()?.daemon.supervisorGeneration).toBe( + replacementHelloB?.supervisorGeneration, + ); + + const denied = await ownerB.client.request({ + type: "complete_owned_session", + activeSessionId: activeA, + }); + expect(denied).toMatchObject({ + success: false, + errorInfo: { code: "owned_session_owner_mismatch" }, + }); + expect(JSON.stringify(denied)).not.toContain(activeA); + for (const descriptor of readWorkerDescriptors(agentDir)) { + const serialized = JSON.stringify(descriptor); + expect(serialized).not.toContain(ownerA.canary); + expect(serialized).not.toContain(ownerB.canary); + expect(descriptor.callerOwnedEnvironmentContract).toBe(true); + expect(descriptor.createCommand).not.toHaveProperty("launchEnv"); + expect(descriptor.createCommand).not.toHaveProperty("launchEnvMode"); + } + const daemonLogs = readDaemonLogs(agentDir); + expect(daemonLogs).not.toContain(ownerA.canary); + expect(daemonLogs).not.toContain(ownerB.canary); + expect(JSON.stringify(ownerA.connection.getOwnedSessionContractProof())).not.toContain(ownerA.canary); + + const [cleanupA, cleanupB] = await Promise.all([ + ownerA.connection.disposeOwnedSession({ timeoutMs: 30_000 }), + ownerB.connection.disposeOwnedSession({ timeoutMs: 30_000 }), + ]); + expect(cleanupA.status).toBe("completed"); + expect(cleanupB.status).toBe("completed"); + expect(JSON.stringify([cleanupA, cleanupB])).not.toContain("issue33-private"); + await waitForProcessGone(recoveredA.pid); + await waitForProcessGone(originalB.pid); + identityTrackedProcesses.delete(recoveredA.pid); + identityTrackedProcesses.delete(originalB.pid); + const shutdown = await ownerA.client.request({ type: "shutdown" }); + if (!shutdown.success) throw new Error("Replacement supervisor rejected test shutdown"); + ownerA.client.close(); + ownerB.client.close(); + await waitForProcessGone(replacementHelloA.supervisorPid); + identityTrackedProcesses.delete(replacementHelloA.supervisorPid); + if (replacementSupervisor.exitCode === null && replacementSupervisor.signalCode === null) { + replacementSupervisor.kill("SIGTERM"); + } + await waitForExit(replacementSupervisor); + children.delete(replacementSupervisor); + + // Detached launchers admitted during the killed-supervisor window can finish + // after the tracked replacement exits. Drain only supervisors authenticated + // on this test-owned socket, and require two clients to agree on identity. + let unavailableSince = Date.now(); + const drainDeadline = Date.now() + 5000; + while (Date.now() < drainDeadline) { + const first = new DaemonClient(socketPath); + try { + await first.connect(100); + } catch { + first.close(); + if (Date.now() - unavailableSince >= 1000) break; + await new Promise((resolveDelay) => setTimeout(resolveDelay, 50)); + continue; + } + unavailableSince = Date.now(); + const second = new DaemonClient(socketPath); + try { + await second.connect(1000); + const firstHello = await first.waitForHello(1000); + const secondHello = await second.waitForHello(1000); + if ( + !firstHello.supervisorPid || + !firstHello.supervisorProcessStartId || + secondHello.supervisorPid !== firstHello.supervisorPid || + secondHello.supervisorProcessStartId !== firstHello.supervisorProcessStartId || + getProcessStartId(firstHello.supervisorPid) !== firstHello.supervisorProcessStartId + ) { + throw new Error("Late test supervisor identity was not current and shared"); + } + identityTrackedProcesses.set(firstHello.supervisorPid, firstHello.supervisorProcessStartId); + const lateShutdown = await first.request({ type: "shutdown" }, 2000); + if (!lateShutdown.success) throw new Error("Late test supervisor rejected shutdown"); + await waitForProcessGone(firstHello.supervisorPid); + identityTrackedProcesses.delete(firstHello.supervisorPid); + } finally { + first.close(); + second.close(); + } + } + if (Date.now() - unavailableSince < 1000) { + throw new Error("Test-owned supervisor socket did not remain unavailable"); + } + }, + 75_000, + ); + + it.runIf(process.platform === "win32")( + "proves exact owned cleanup over a Windows named pipe", + async () => { + const root = tempDir(); + const agentDir = join(root, "agent"); + const projectDir = join(root, "project"); + const sessionDir = join(agentDir, "sessions"); + const sessionFile = createSnapshotSessionFile(agentDir, projectDir, "issue #33 Windows owner"); + const socketPath = String.raw`\\.\pipe\prime-issue33-${process.pid}-${randomUUID().slice(0, 8)}`; + mkdirSync(projectDir, { recursive: true }); + const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); + const owner = await connectEventually(socketPath, supervisor); + const launchEnv = collectDaemonLaunchEnv(process.env); + for (const name of Object.keys(launchEnv)) { + if (name.startsWith("RLM_") || /TOKEN|KEY|SECRET|PASSWORD|CREDENTIAL|AUTH|COOKIE/i.test(name)) { + delete launchEnv[name]; + } + } + Object.assign(launchEnv, { + PI_OFFLINE: "1", + TSX_TSCONFIG_PATH: resolve(__dirname, "../../../tsconfig.json"), + [ENV_AGENT_DIR]: agentDir, + PRIME_AGENT_TEST_WINDOWS_ENV_A: "windows-a", + }); + const created = await owner.request({ + type: "create", + lifecycle: "client_owned", + sessionPath: sessionFile, + launchEnv, + launchEnvMode: "replace", + config: { + cwd: projectDir, + agentDir, + sessionDir, + provider: "faux", + model: "faux", + noTools: true, + noExtensions: true, + }, + }); + if (!created.success) throw new Error(created.error); + const summary = requireSummary(created.data); + if (!summary.workerPid) throw new Error("Windows owned worker did not expose its pid"); + const activeSessionId = summary.activeSessionId ?? summary.id; + const descriptor = readWorkerDescriptors(agentDir).find( + (candidate) => candidate.rootActiveSessionId === activeSessionId, + ); + if (!descriptor?.pid || !descriptor.processStartId) { + throw new Error("Windows owned worker did not expose a complete process identity"); + } + identityTrackedProcesses.set(descriptor.pid, descriptor.processStartId); + const connection = await DaemonAgentConnection.attach(owner, activeSessionId, { + ownedSession: true, + ownedSessionLaunchEnv: launchEnv, + ownedSessionRecoveryConfig: { + cwd: projectDir, + agentDir, + sessionDir, + provider: "faux", + model: "faux", + noTools: true, + noExtensions: true, + }, + supportsExtensionUi: false, + }); + expect(connection.getOwnedSessionContractProof()).toMatchObject({ + feature: "caller_owned_session_environment_cleanup_v1", + status: "attached", + }); + const cleanup = await connection.disposeOwnedSession({ timeoutMs: 30_000 }); + expect(cleanup.status).toBe("completed"); + await waitForProcessGone(descriptor.pid); + identityTrackedProcesses.delete(descriptor.pid); + await owner.request({ type: "shutdown" }); + owner.close(); + await waitForSocketGone(socketPath); + await waitForExit(supervisor); + }, + 60_000, + ); + it("lets a different client prove cleanup after the owner socket disappears", async () => { const root = tempDir(); const agentDir = join(root, "agent"); @@ -1147,7 +1612,6 @@ describe("daemon supervisor resident workers", () => { const observer = await connectEventually(socketPath, supervisor); try { - let sawStopping = false; let settled = false; const deadline = Date.now() + 50_000; while (Date.now() < deadline) { @@ -1157,14 +1621,14 @@ describe("daemon supervisor resident workers", () => { }); if (!response.success) throw new Error(response.error); const status = (response.data as { status?: string } | undefined)?.status; - if (status === "stopping") sawStopping = true; if (status === "settled") { settled = true; break; } await new Promise((resolveDelay) => setTimeout(resolveDelay, 50)); } - expect(sawStopping).toBe(true); + // `settled` is the authoritative result; the observer may miss the + // transient `stopping` phase when registration and cleanup race. expect(settled).toBe(true); await waitForProcessGone(descriptor.pid); workerPids.delete(descriptor.pid); diff --git a/packages/coding-agent/test/suite/regressions/33-caller-owned-session-contract.test.ts b/packages/coding-agent/test/suite/regressions/33-caller-owned-session-contract.test.ts new file mode 100644 index 0000000000..eaa72b22d5 --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/33-caller-owned-session-contract.test.ts @@ -0,0 +1,34 @@ +import { fauxAssistantMessage } from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it } from "vitest"; +import { PRIME_AGENT_SDK_FEATURES } from "../../../src/index.js"; +import { cloneCallerOwnedSessionLaunchEnv } from "../../../src/modes/daemon/daemon-protocol.js"; +import { createHarness, getAssistantTexts, type Harness } from "../harness.js"; + +describe("issue #33 caller-owned session contract", () => { + let harness: Harness | undefined; + + afterEach(() => { + harness?.cleanup(); + harness = undefined; + }); + + it("freezes the launch snapshot without leaking it into an ordinary faux-provider turn", async () => { + const source = { PROVIDER_TOKEN: "issue-33-private-a", PATH: "/caller/a" }; + const snapshot = cloneCallerOwnedSessionLaunchEnv(source); + source.PROVIDER_TOKEN = "issue-33-private-c"; + source.PATH = "/ambient/c"; + + harness = await createHarness(); + harness.setResponses([fauxAssistantMessage("contract remains transport-only")]); + await harness.session.prompt("run a faux-provider turn"); + + expect(snapshot).toEqual({ PROVIDER_TOKEN: "issue-33-private-a", PATH: "/caller/a" }); + expect(Object.isFrozen(snapshot)).toBe(true); + expect(PRIME_AGENT_SDK_FEATURES).toContain("caller_owned_session_environment_cleanup_v1"); + expect(getAssistantTexts(harness)).toContain("contract remains transport-only"); + const sessionSurface = JSON.stringify(harness.events); + expect(sessionSurface).not.toContain("issue-33-private"); + expect(sessionSurface).not.toContain("/caller/a"); + expect(sessionSurface).not.toContain("/ambient/c"); + }); +});