diff --git a/packages/opencode/src/actor/spawn.ts b/packages/opencode/src/actor/spawn.ts index 734672466..3aee91b5e 100644 --- a/packages/opencode/src/actor/spawn.ts +++ b/packages/opencode/src/actor/spawn.ts @@ -195,7 +195,7 @@ export interface SpawnResult { export interface Interface { readonly spawn: (input: SpawnInput) => Effect.Effect readonly cancel: (sessionID: SessionID, actorID: string, mode: "graceful" | "forced") => Effect.Effect - readonly getForkContext: (actorID: string) => Effect.Effect + readonly getForkContext: (sessionID: SessionID, actorID: string) => Effect.Effect /** * Run ONE stall-watchdog scan pass synchronously (the same body the background * fiber repeats every WATCHDOG_SCAN_INTERVAL_MS). Exposed for deterministic @@ -226,7 +226,14 @@ export const layer = Layer.effect( // ForkContext snapshot per actor, captured at spawn for fork agents // (contextMode = "full"). Read by fork's runLoop (see prompt.ts) and // cleared on terminal status. Fiber tracking moved to SessionRunState. + // + // Keyed by (sessionID, actorID) rather than actorID alone: actorID is only + // unique within a (sessionID, agentType) scope (see + // ActorRegistry.allocateActorID), so two concurrent sessions spawning the + // same agentType can allocate the same actorID. A bare actorID key would + // let one session's forkContext clobber or be deleted by another's. const forkContexts = new Map() + const forkContextKey = (sessionID: SessionID, actorID: string) => `${sessionID}:${actorID}` // Actors whose cancel() has begun, keyed "sessionID:actorID". Populated // BEFORE the fiber is interrupted so forkWork's onSuccess/onFailure notify @@ -636,7 +643,7 @@ export const layer = Layer.effect( lastFinalText = newTurn.finalText } - yield* Effect.sync(() => forkContexts.delete(input.actorID)) + yield* Effect.sync(() => forkContexts.delete(forkContextKey(input.sessionID, input.actorID))) }), onFailure: (cause) => Effect.gen(function* () { @@ -647,7 +654,7 @@ export const layer = Layer.effect( outcome, cancelled ? { status: "cancelled" as const } : { status: "failure" as const, error }, ) - yield* Effect.sync(() => forkContexts.delete(input.actorID)) + yield* Effect.sync(() => forkContexts.delete(forkContextKey(input.sessionID, input.actorID))) }), }), ) @@ -701,7 +708,7 @@ export const layer = Layer.effect( tools: input.tools, }) if (input.forkContext) { - forkContexts.set(child.id, input.forkContext) // peer's actorID === child.id + forkContexts.set(forkContextKey(child.id, child.id), input.forkContext) // peer's actorID === child.id === sessionID } const { fiber, outcome } = yield* forkWork({ sessionID: child.id, @@ -747,7 +754,7 @@ export const layer = Layer.effect( if (input.onActorID) yield* Effect.sync(() => input.onActorID!(actorID)).pipe(Effect.ignore) if (input.forkContext) { - forkContexts.set(actorID, input.forkContext) + forkContexts.set(forkContextKey(input.sessionID, actorID), input.forkContext) } // Auto-inject return-format instruction for non-specialized subagents. @@ -853,11 +860,11 @@ export const layer = Layer.effect( .updateStatus(sessionID, actorID, { status: "idle", lastOutcome: "cancelled" }) .pipe(Effect.ignore) yield* notifyTerminal(sessionID, actorID, actor, "cancelled") - yield* Effect.sync(() => forkContexts.delete(actorID)) + yield* Effect.sync(() => forkContexts.delete(forkContextKey(sessionID, actorID))) }) - const getForkContext = Effect.fn("Actor.getForkContext")(function* (actorID: string) { - return forkContexts.get(actorID) + const getForkContext = Effect.fn("Actor.getForkContext")(function* (sessionID: SessionID, actorID: string) { + return forkContexts.get(forkContextKey(sessionID, actorID)) }) // === T40 stall watchdog === diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index dc5c1608a..177e4a4be 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/packages/opencode/src/session/prompt.ts @@ -3339,7 +3339,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the // If forkCtx is missing (race / cleanup bug / spawn skipped), fail the // actor so the next prune turn can spawn a fresh fork. if (isForkAgent) { - const forkCtxEffect = spawnRef.current?.getForkContext(lastUser.agentID!) + const forkCtxEffect = spawnRef.current?.getForkContext(sessionID, lastUser.agentID!) const forkCtx = forkCtxEffect ? yield* forkCtxEffect : undefined if (!forkCtx) { yield* slog.warn("fork agent runLoop: missing forkContext, failing actor", { diff --git a/packages/opencode/test/actor/spawn.test.ts b/packages/opencode/test/actor/spawn.test.ts index 388da8ce0..b119d3b11 100644 --- a/packages/opencode/test/actor/spawn.test.ts +++ b/packages/opencode/test/actor/spawn.test.ts @@ -700,7 +700,7 @@ describe("Actor forkContext lifecycle", () => { }) // Before cancel: forkContext must be present. - const before = yield* actor.getForkContext(result.actorID) + const before = yield* actor.getForkContext(result.sessionID, result.actorID) expect(before).toBeDefined() expect(before?.system).toEqual(["test-system"]) @@ -708,7 +708,7 @@ describe("Actor forkContext lifecycle", () => { yield* actor.cancel(result.sessionID, result.actorID, "forced") // After cancel: forkContext must be gone. - const after = yield* actor.getForkContext(result.actorID) + const after = yield* actor.getForkContext(result.sessionID, result.actorID) expect(after).toBeUndefined() }), { git: true, config: providerCfg }, @@ -751,7 +751,7 @@ describe("mode × contextMode matrix", () => { forkContext: fakeForkCtx, }) - const ctx = yield* actor.getForkContext(result.actorID) + const ctx = yield* actor.getForkContext(result.sessionID, result.actorID) expect(ctx).toBeDefined() expect(ctx?.system).toEqual(["test-system"]) @@ -761,6 +761,90 @@ describe("mode × contextMode matrix", () => { ), ) + it.live("forkContext does not leak across sessions that allocate the same actorID", () => + provideTmpdirServer( + Effect.fnUntraced(function* ({ llm }) { + const actor = yield* Actor.Service + const session = yield* Session.Service + + // Two independent sessions. allocateActorID scopes its incrementing + // suffix by (sessionID, agentType), so the first "explore" subagent + // spawned in each session below is expected to collide on the same + // actorID (e.g. "explore-1") even though the sessions are unrelated. + const sessionA = yield* session.create({ + title: "forkCtx cross-session A", + permission: [{ permission: "*", pattern: "*", action: "allow" }], + }) + const sessionB = yield* session.create({ + title: "forkCtx cross-session B", + permission: [{ permission: "*", pattern: "*", action: "allow" }], + }) + + // Hang the LLM so both fibers stay alive while we verify state. + yield* llm.hang + + const forkCtxA = { + system: ["session-a"], + tools: {}, + inheritedMessages: [], + parentPermission: [], + watermarkMsgID: MessageID.ascending(), + model: ref, + } + const forkCtxB = { + system: ["session-b"], + tools: {}, + inheritedMessages: [], + parentPermission: [], + watermarkMsgID: MessageID.ascending(), + model: ref, + } + + const resultA = yield* actor.spawn({ + mode: "subagent", + sessionID: sessionA.id, + agentType: "explore", + task: "noop", + context: "none", + tools: [], + background: true, + model: ref, + forkContext: forkCtxA, + }) + const resultB = yield* actor.spawn({ + mode: "subagent", + sessionID: sessionB.id, + agentType: "explore", + task: "noop", + context: "none", + tools: [], + background: true, + model: ref, + forkContext: forkCtxB, + }) + + // Confirms the collision precondition: both sessions allocated the + // same locally-scoped actorID. + expect(resultA.actorID).toBe(resultB.actorID) + + // Each session must read back its own forkContext, not the other's. + const ctxA = yield* actor.getForkContext(sessionA.id, resultA.actorID) + const ctxB = yield* actor.getForkContext(sessionB.id, resultB.actorID) + expect(ctxA?.system).toEqual(["session-a"]) + expect(ctxB?.system).toEqual(["session-b"]) + + yield* actor.cancel(sessionA.id, resultA.actorID, "forced") + + // Cancelling session A's actor must not clear session B's forkContext. + const ctxBAfterCancelA = yield* actor.getForkContext(sessionB.id, resultB.actorID) + expect(ctxBAfterCancelA?.system).toEqual(["session-b"]) + + yield* actor.cancel(sessionB.id, resultB.actorID, "forced") + }), + { git: true, config: providerCfg }, + ), + ) + it.live("subagent + none: no forkContext stored", () => provideTmpdirServer( Effect.fnUntraced(function* ({ llm }) { @@ -786,7 +870,7 @@ describe("mode × contextMode matrix", () => { // no forkContext }) - const ctx = yield* actor.getForkContext(result.actorID) + const ctx = yield* actor.getForkContext(result.sessionID, result.actorID) expect(ctx).toBeUndefined() yield* actor.cancel(result.sessionID, result.actorID, "forced") @@ -822,7 +906,7 @@ describe("mode × contextMode matrix", () => { // For peer, result.actorID === child.id (the new session id) expect(result.actorID).not.toBe(parent.id) - const ctx = yield* actor.getForkContext(result.actorID) + const ctx = yield* actor.getForkContext(result.sessionID, result.actorID) expect(ctx).toBeDefined() expect(ctx?.system).toEqual(["test-system"]) @@ -857,7 +941,7 @@ describe("mode × contextMode matrix", () => { // no forkContext }) - const ctx = yield* actor.getForkContext(result.actorID) + const ctx = yield* actor.getForkContext(result.sessionID, result.actorID) expect(ctx).toBeUndefined() yield* actor.cancel(result.sessionID, result.actorID, "forced") diff --git a/packages/opencode/test/inbox/fork-agent-compat.test.ts b/packages/opencode/test/inbox/fork-agent-compat.test.ts index 2d7d16528..61b4fa129 100644 --- a/packages/opencode/test/inbox/fork-agent-compat.test.ts +++ b/packages/opencode/test/inbox/fork-agent-compat.test.ts @@ -315,7 +315,7 @@ describe("Fork-agent inbox compat (Plan 4 / Task 5)", () => { const forkActorID = result.actorID // Tier 2 — structural: capture forkContext BEFORE inbox send. - const forkCtxBefore = yield* actor.getForkContext(forkActorID) + const forkCtxBefore = yield* actor.getForkContext(result.sessionID, forkActorID) expect(forkCtxBefore).toBeDefined() expect(forkCtxBefore?.system).toEqual(["inherited-system-prompt"]) expect(forkCtxBefore?.inheritedMessages).toHaveLength(2) @@ -356,7 +356,7 @@ describe("Fork-agent inbox compat (Plan 4 / Task 5)", () => { // Tier 2 — structural: forkContext must still have the exact same // inheritedMessages snapshot. Drain must not touch forkContexts. - const forkCtxAfter = yield* actor.getForkContext(forkActorID) + const forkCtxAfter = yield* actor.getForkContext(result.sessionID, forkActorID) expect(forkCtxAfter).toBeDefined() expect(forkCtxAfter?.inheritedMessages).toStrictEqual(forkCtxBefore?.inheritedMessages) expect(forkCtxAfter?.system).toEqual(["inherited-system-prompt"]) diff --git a/packages/opencode/test/session/bootstrap-skip-system.test.ts b/packages/opencode/test/session/bootstrap-skip-system.test.ts index 88cfc989c..380411687 100644 --- a/packages/opencode/test/session/bootstrap-skip-system.test.ts +++ b/packages/opencode/test/session/bootstrap-skip-system.test.ts @@ -53,7 +53,7 @@ const stubActor = Layer.succeed( Actor.Service.of({ spawn: () => Effect.die("Actor.spawn unexpectedly called in system-spawn skip path"), cancel: () => Effect.die("Actor.cancel unexpectedly called in system-spawn skip path"), - getForkContext: () => Effect.succeed(undefined), + getForkContext: (_sessionID: string, _actorID: string) => Effect.succeed(undefined), }), ) diff --git a/packages/opencode/test/session/checkpoint-child-session.test.ts b/packages/opencode/test/session/checkpoint-child-session.test.ts index 171391c35..d869c22db 100644 --- a/packages/opencode/test/session/checkpoint-child-session.test.ts +++ b/packages/opencode/test/session/checkpoint-child-session.test.ts @@ -76,7 +76,7 @@ const recordingActor = Layer.effect( } }), cancel: () => Effect.void, - getForkContext: () => Effect.succeed(undefined), + getForkContext: (_sessionID: string, _actorID: string) => Effect.succeed(undefined), }) spawnRef.current = impl yield* Effect.addFinalizer(() => diff --git a/packages/opencode/test/session/checkpoint-drain.test.ts b/packages/opencode/test/session/checkpoint-drain.test.ts index 2eece2c84..ea43294a5 100644 --- a/packages/opencode/test/session/checkpoint-drain.test.ts +++ b/packages/opencode/test/session/checkpoint-drain.test.ts @@ -48,7 +48,7 @@ const hangingActor = Layer.effect( } }), cancel: () => Effect.void, - getForkContext: () => Effect.succeed(undefined), + getForkContext: (_sessionID: string, _actorID: string) => Effect.succeed(undefined), }) spawnRef.current = impl yield* Effect.addFinalizer(() => diff --git a/packages/opencode/test/session/checkpoint-fork-mode.test.ts b/packages/opencode/test/session/checkpoint-fork-mode.test.ts index ced893b8e..a47acd986 100644 --- a/packages/opencode/test/session/checkpoint-fork-mode.test.ts +++ b/packages/opencode/test/session/checkpoint-fork-mode.test.ts @@ -59,7 +59,7 @@ const recordingActor = Layer.effect( } }), cancel: () => Effect.void, - getForkContext: () => Effect.succeed(undefined), + getForkContext: (_sessionID: string, _actorID: string) => Effect.succeed(undefined), }) spawnRef.current = impl yield* Effect.addFinalizer(() => diff --git a/packages/opencode/test/session/checkpoint-main-slice.test.ts b/packages/opencode/test/session/checkpoint-main-slice.test.ts index b5beb492b..00b4e7ec3 100644 --- a/packages/opencode/test/session/checkpoint-main-slice.test.ts +++ b/packages/opencode/test/session/checkpoint-main-slice.test.ts @@ -58,7 +58,7 @@ const recordingActor = Layer.effect( } }), cancel: () => Effect.void, - getForkContext: () => Effect.succeed(undefined), + getForkContext: (_sessionID: string, _actorID: string) => Effect.succeed(undefined), }) spawnRef.current = impl diff --git a/packages/opencode/test/session/checkpoint-rebuild-fallback-decision.test.ts b/packages/opencode/test/session/checkpoint-rebuild-fallback-decision.test.ts index 561e4ac0a..6b9cf4b24 100644 --- a/packages/opencode/test/session/checkpoint-rebuild-fallback-decision.test.ts +++ b/packages/opencode/test/session/checkpoint-rebuild-fallback-decision.test.ts @@ -43,7 +43,7 @@ const controllableActor = Layer.effect( return { actorID: `${input.agentType}-${counter}`, sessionID: input.sessionID, outcome } }), cancel: () => Effect.void, - getForkContext: () => Effect.succeed(undefined), + getForkContext: (_sessionID: string, _actorID: string) => Effect.succeed(undefined), }) spawnRef.current = impl yield* Effect.addFinalizer(() => diff --git a/packages/opencode/test/session/checkpoint-rebuild-nonblocking.test.ts b/packages/opencode/test/session/checkpoint-rebuild-nonblocking.test.ts index 7f74c2a94..e6c715eb4 100644 --- a/packages/opencode/test/session/checkpoint-rebuild-nonblocking.test.ts +++ b/packages/opencode/test/session/checkpoint-rebuild-nonblocking.test.ts @@ -48,7 +48,7 @@ const hangingActor = Layer.effect( } }), cancel: () => Effect.void, - getForkContext: () => Effect.succeed(undefined), + getForkContext: (_sessionID: string, _actorID: string) => Effect.succeed(undefined), }) spawnRef.current = impl yield* Effect.addFinalizer(() => diff --git a/packages/opencode/test/session/checkpoint-watermark-transactional.test.ts b/packages/opencode/test/session/checkpoint-watermark-transactional.test.ts index b5b59658a..0b0ff467c 100644 --- a/packages/opencode/test/session/checkpoint-watermark-transactional.test.ts +++ b/packages/opencode/test/session/checkpoint-watermark-transactional.test.ts @@ -45,7 +45,7 @@ const controllableActor = Layer.effect( return { actorID: `${input.agentType}-${counter}`, sessionID: input.sessionID, outcome } }), cancel: () => Effect.void, - getForkContext: () => Effect.succeed(undefined), + getForkContext: (_sessionID: string, _actorID: string) => Effect.succeed(undefined), }) spawnRef.current = impl yield* Effect.addFinalizer(() => diff --git a/packages/opencode/test/session/classify-integration.test.ts b/packages/opencode/test/session/classify-integration.test.ts index 57f349bfa..83305263e 100644 --- a/packages/opencode/test/session/classify-integration.test.ts +++ b/packages/opencode/test/session/classify-integration.test.ts @@ -368,7 +368,8 @@ describe("classifier routing — integration", () => { model: { providerID: ProviderID.make("alibaba"), modelID: ModelID.make("qwen-plus") }, } spawnRef.current = { - getForkContext: (id: string) => Effect.succeed(id === forkActorID ? forkCtx : undefined), + getForkContext: (sessionID: string, actorID: string) => + Effect.succeed(actorID === forkActorID ? forkCtx : undefined), spawn: () => Effect.die("spawn not used in fork-gate test"), cancel: () => Effect.die("cancel not used in fork-gate test"), } as unknown as NonNullable @@ -443,7 +444,8 @@ describe("classifier routing — integration", () => { model: { providerID: ProviderID.make("alibaba"), modelID: ModelID.make("qwen-plus") }, } spawnRef.current = { - getForkContext: (id: string) => Effect.succeed(id === forkActorID ? forkCtx : undefined), + getForkContext: (sessionID: string, actorID: string) => + Effect.succeed(actorID === forkActorID ? forkCtx : undefined), spawn: () => Effect.die("spawn not used in fork content-filter test"), cancel: () => Effect.die("cancel not used in fork content-filter test"), } as unknown as NonNullable diff --git a/packages/opencode/test/tool/actor-cancel.test.ts b/packages/opencode/test/tool/actor-cancel.test.ts index 77020af4c..f512055dc 100644 --- a/packages/opencode/test/tool/actor-cancel.test.ts +++ b/packages/opencode/test/tool/actor-cancel.test.ts @@ -71,7 +71,7 @@ beforeAll(() => { yield* installedRegistry.updateStatus(sessionID, actorID, { status: "idle", lastOutcome: "cancelled" }).pipe(Effect.ignore) } }), - getForkContext: () => Effect.succeed(undefined), + getForkContext: (_sessionID: string, _actorID: string) => Effect.succeed(undefined), } satisfies ActorInterface }) afterAll(() => { diff --git a/packages/opencode/test/tool/actor.test.ts b/packages/opencode/test/tool/actor.test.ts index e5308d1cd..37f5c2caf 100644 --- a/packages/opencode/test/tool/actor.test.ts +++ b/packages/opencode/test/tool/actor.test.ts @@ -70,7 +70,7 @@ function installMockSpawn(onSpawn?: (input: SpawnInput) => void) { return { actorID, sessionID: input.sessionID, outcome } }), cancel: () => Effect.void, - getForkContext: () => Effect.succeed(undefined), + getForkContext: (_sessionID: string, _actorID: string) => Effect.succeed(undefined), } }) }