From 8c518da60367e2e55b4eb5d5dd9562fc6b2bf258 Mon Sep 17 00:00:00 2001 From: owen Date: Sun, 6 Sep 2026 05:17:04 +0000 Subject: [PATCH 1/5] feat(server): suspend idle Pi runtimes and notify clients Kill held Pi processes after an idle timeout without sealing the session, publish session.runtime.stopped, and let the next prompt ensureRuntime again. UI clears live turn state but stays ready (not a terminal error). --- CONTEXT.md | 3 + .../src/features/chat/runtime/chat.test.ts | 13 +++ apps/app/src/features/chat/runtime/chat.ts | 11 +++ packages/contract/src/domain.ts | 9 +- .../server/src/harness/events/framework.ts | 5 + packages/server/src/harness/runtime-idle.ts | 16 ++++ packages/server/src/harness/session-fold.ts | 13 +++ packages/server/src/harness/session.ts | 92 ++++++++++++++++++- .../test/harness/event-manifest.test.ts | 3 +- packages/server/test/harness/session.test.ts | 62 +++++++++++++ .../harness/types/session-envelope.test.ts | 1 + 11 files changed, 224 insertions(+), 4 deletions(-) create mode 100644 packages/server/src/harness/runtime-idle.ts diff --git a/CONTEXT.md b/CONTEXT.md index ac17f00da..a996ee466 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -47,6 +47,9 @@ Effect Context service: availability check, create/resume, and cold reads. Const **PiAgentRuntime / PiProcess** (`harness/pi/runtime.ts`, `harness/pi/process.ts`): `PiAgentRuntime` is the live execution resource (prompt/events/close) for one agent session id. `PiProcess` spawns and owns the underlying `pi --mode rpc` child. +**Runtime idle suspend**: +A held Pi runtime that stays at phase `idle` for `PIE_SESSION_RUNTIME_IDLE_MS` (default 5 minutes; `0` disables) is suspended: the process is killed without sealing the session, and `session.runtime.stopped` is published. The next prompt re-`ensureRuntime`s. Clients clear live turn UI but do not treat this as terminal error. + **Private modules** (no Context tags, never wired directly): `harness/session.ts` — **PiAgentSession**, one session as this server sees it: seq stamping, phase, buffers, pending requests, and the single-flight lifecycle of the runtime it _optionally_ owns. `harness/session-fold.ts` — the pure state fold. `harness/session-repository.ts` — metadata store over `storage/sessions/`. diff --git a/apps/app/src/features/chat/runtime/chat.test.ts b/apps/app/src/features/chat/runtime/chat.test.ts index a5da5de75..6b0e2d469 100644 --- a/apps/app/src/features/chat/runtime/chat.test.ts +++ b/apps/app/src/features/chat/runtime/chat.test.ts @@ -843,6 +843,19 @@ describe("Chat truncated buffers", () => { }); describe("Chat lifecycle", () => { + it("treats runtime.stopped as idle/ready, not a terminal error", async () => { + const { chat, live, attach } = makeChat(); + await attach({}); + live(1, { type: "session.turn.started", turnId: "t1", phase: "running" }); + live(2, { + type: "session.runtime.stopped", + reason: "idle", + phase: "idle", + }); + expect(chat.store.getState().status).toBe("ready"); + expect(chat.store.getState().error).toBeUndefined(); + }); + it("copies the crashed phase into an error status", async () => { const { chat, attach, live } = makeChat(); await attach({}); diff --git a/apps/app/src/features/chat/runtime/chat.ts b/apps/app/src/features/chat/runtime/chat.ts index 129d40775..dfee62182 100644 --- a/apps/app/src/features/chat/runtime/chat.ts +++ b/apps/app/src/features/chat/runtime/chat.ts @@ -232,6 +232,17 @@ export class Chat { this.#state.clearPendingRequests(); this.#state.clearPendingPrompt(); break; + // Server killed an idle Pi process; session stays. Clear live turn UI + // the same way as a crash, but phase stays idle → composer ready. + // The EventBus subscription keeps flowing; the next prompt re-spawns + // the process via ensureRuntime — no client re-attach required. + case "session.runtime.stopped": + for (const fold of this.#turnFolds.values()) fold.close(); + this.#turnFolds.clear(); + this.#state.clearPendingRequests(); + this.#state.clearPendingPrompt(); + this.#state.retryNotice = undefined; + break; } // Status is copied off the event (the runtime stamps its post-event // phase), never derived from event types here. Lifecycle events only: diff --git a/packages/contract/src/domain.ts b/packages/contract/src/domain.ts index c97e946d5..4934b5f6b 100644 --- a/packages/contract/src/domain.ts +++ b/packages/contract/src/domain.ts @@ -196,6 +196,9 @@ export const SessionScopedEventTypes = [ "session.request.rejected", "session.queue.updated", "session.crashed", + // Runtime was stopped on purpose (e.g. idle timeout). Session stays; + // the next prompt re-acquires a process. Phase is idle — not crashed. + "session.runtime.stopped", ] as const; export type SessionScopedEventType = (typeof SessionScopedEventTypes)[number]; @@ -257,7 +260,11 @@ export type SessionScopedEventBody = readonly steering: ReadonlyArray; readonly followUp: ReadonlyArray; } - | { readonly type: "session.crashed"; readonly reason: string }; + | { readonly type: "session.crashed"; readonly reason: string } + // Process gone, session kept. `reason` is a short machine token + // (`idle` today). Clients clear live turn UI but must not treat this + // as a terminal error — prompting will resume the runtime. + | { readonly type: "session.runtime.stopped"; readonly reason: string }; /** A session-scoped event before the server's `HarnessAgentSession` stamps its `seq`. */ export type SessionScopedEventDraft = { readonly ref: SessionRef } & SessionScopedEventBody; diff --git a/packages/server/src/harness/events/framework.ts b/packages/server/src/harness/events/framework.ts index 2e00361aa..3796d8af7 100644 --- a/packages/server/src/harness/events/framework.ts +++ b/packages/server/src/harness/events/framework.ts @@ -65,6 +65,10 @@ export const SessionCrashed = defineEvent({ type: "session.crashed", schema: { ...sid, reason: Schema.String }, }); +export const SessionRuntimeStopped = defineEvent({ + type: "session.runtime.stopped", + schema: { ...sid, reason: Schema.String }, +}); export const SessionCreated = defineEvent({ type: "session.created", schema: { sessionId: Schema.String }, @@ -93,6 +97,7 @@ export const SessionEventDefs = [ SessionRequestReplied, SessionQueueUpdated, SessionCrashed, + SessionRuntimeStopped, ] as const; export type SessionEvent = EventValue<(typeof SessionEventDefs)[number]>; diff --git a/packages/server/src/harness/runtime-idle.ts b/packages/server/src/harness/runtime-idle.ts new file mode 100644 index 000000000..63f7ab728 --- /dev/null +++ b/packages/server/src/harness/runtime-idle.ts @@ -0,0 +1,16 @@ +/** + * How long a live Pi runtime may sit at phase `idle` before the session + * suspends it (kills the process, keeps the session). Overridable for tests + * and ops via `PIE_SESSION_RUNTIME_IDLE_MS` (milliseconds). `0` disables. + */ +export const DEFAULT_SESSION_RUNTIME_IDLE_MS = 5 * 60 * 1000; + +export const sessionRuntimeIdleMs = ( + env: NodeJS.ProcessEnv = process.env, +): number => { + const raw = env.PIE_SESSION_RUNTIME_IDLE_MS?.trim(); + if (raw === undefined || raw === "") return DEFAULT_SESSION_RUNTIME_IDLE_MS; + const parsed = Number(raw); + if (!Number.isFinite(parsed) || parsed < 0) return DEFAULT_SESSION_RUNTIME_IDLE_MS; + return Math.floor(parsed); +}; diff --git a/packages/server/src/harness/session-fold.ts b/packages/server/src/harness/session-fold.ts index 58cfb3a2f..2e73aa39e 100644 --- a/packages/server/src/harness/session-fold.ts +++ b/packages/server/src/harness/session-fold.ts @@ -129,6 +129,8 @@ export const toWireBody = ( }; case "session.crashed": return { type: "session.crashed", reason: event.reason }; + case "session.runtime.stopped": + return { type: "session.runtime.stopped", reason: event.reason }; } }; @@ -251,6 +253,17 @@ export const foldSessionEvent = ( pendingRequests: new Map(), pendingPrompt: emptyPendingPrompt, }; + case "session.runtime.stopped": + // Intentional stop (idle timeout): back to idle, clear live turn + // state. seq/cursor keep going so clients do not rewind. + return { + ...base, + phase: "idle", + activeTurn: null, + activePrompt: null, + pendingRequests: new Map(), + pendingPrompt: emptyPendingPrompt, + }; } }; diff --git a/packages/server/src/harness/session.ts b/packages/server/src/harness/session.ts index b0df20f0f..509c049ec 100644 --- a/packages/server/src/harness/session.ts +++ b/packages/server/src/harness/session.ts @@ -10,6 +10,7 @@ import { Deferred, Effect, Exit, Fiber, Ref, Scope, Semaphore, Stream } from "ef import type { EventBusShape } from "../events/event-bus"; import type { ResumeSessionError } from "./errors"; import type { PiAgentRuntime } from "./pi/runtime"; +import { sessionRuntimeIdleMs } from "./runtime-idle"; import { foldSessionEvent, initialSessionState, @@ -126,20 +127,37 @@ export type PiAgentSessionShape = { * rather than racing it. */ readonly releaseRuntime: Effect.Effect; + /** + * Kill the held Pi runtime without sealing the session, and publish + * `session.runtime.stopped`. The next {@link ensureRuntime} may resume. + * No-op when nothing is held. Used for idle timeout. + */ + readonly suspendRuntime: (reason?: string) => Effect.Effect; +}; + +export type MakePiAgentSessionOptions = { + /** Idle kill delay in ms. `0` disables. Defaults to {@link sessionRuntimeIdleMs}. */ + readonly idleTimeoutMs?: number; }; export const makePiAgentSession = ( ref: SessionRef, bus: EventBusShape, + options: MakePiAgentSessionOptions = {}, ): Effect.Effect => Effect.gen(function* () { const ownerScope = yield* Scope.Scope; + const idleTimeoutMs = options.idleTimeoutMs ?? sessionRuntimeIdleMs(); const state = yield* Ref.make(initialSessionState); const lifecycle = yield* Ref.make({ held: undefined, acquiring: undefined, sealed: false, }); + // Generation bumps cancel an in-flight idle timer without racing suspend. + const idleGeneration = yield* Ref.make(0); + // Filled once idle helpers exist; applyWith calls it after every publish. + let afterPublish: Effect.Effect = Effect.void; // Serializes stamp+publish across the drain fiber and `emit` callers: // without it two fibers could stamp seqs n/n+1 but publish n+1 first, and @@ -215,6 +233,7 @@ export const makePiAgentSession = ( // After the publish: subscribers are the ones waiting on this, // and the log must not sit in front of them. Effect.andThen(logTurn(wireBody, event.seq)), + Effect.andThen(Effect.suspend(() => afterPublish)), ); }), ), @@ -393,6 +412,74 @@ export const makePiAgentSession = ( Effect.uninterruptible, ); + const cancelIdleTimer: Effect.Effect = Ref.update(idleGeneration, (n) => n + 1); + + /** + * Drop the held runtime without sealing, publish `session.runtime.stopped`, + * leave the session queryable so the next prompt can {@link ensureRuntime}. + */ + const suspendRuntime = (reason = "idle"): Effect.Effect => + identified( + cancelIdleTimer.pipe( + Effect.andThen( + Ref.modify(lifecycle, (current) => { + if (current.sealed || !current.held) { + return [undefined as Held | undefined, current] as const; + } + return [current.held, { ...current, held: undefined }] as const; + }), + ), + Effect.flatMap((held) => + held + ? shutDown(held).pipe( + Effect.andThen( + apply({ + type: "session.runtime.stopped", + sessionId: ref.sessionId, + reason, + }), + ), + Effect.andThen( + Effect.logInfo("session runtime suspended").pipe( + Effect.annotateLogs({ event: "session.runtime.stopped", reason }), + ), + ), + ) + : Effect.void, + ), + Effect.uninterruptible, + ), + ); + + /** Arm idle kill when we hold a runtime and the fold says idle. */ + const armIdleTimer: Effect.Effect = Effect.gen(function* () { + if (idleTimeoutMs <= 0) return; + const held = yield* Ref.get(lifecycle).pipe(Effect.map((current) => current.held)); + if (!held) return; + const phase = yield* Ref.get(state).pipe(Effect.map((current) => current.phase)); + if (phase !== "idle") return; + const generation = yield* Ref.updateAndGet(idleGeneration, (n) => n + 1); + yield* Effect.forkIn( + identified( + Effect.sleep(`${idleTimeoutMs} millis`).pipe( + Effect.andThen(Ref.get(idleGeneration)), + Effect.flatMap((current) => + current === generation ? suspendRuntime("idle") : Effect.void, + ), + ), + ), + ownerScope, + ); + }); + + // Wrap ensureRuntime to (re)arm idle once a runtime is held while idle. + const ensureRuntimeWithIdle: PiAgentSessionShape["ensureRuntime"] = (acquire) => + ensureRuntime(acquire).pipe(Effect.tap(() => armIdleTimer)); + + afterPublish = Ref.get(state).pipe( + Effect.flatMap((current) => (current.phase === "idle" ? armIdleTimer : cancelIdleTimer)), + ); + return { ref, snapshot: applyLock.withPermit( @@ -401,7 +488,8 @@ export const makePiAgentSession = ( status: Ref.get(state).pipe(Effect.map(toStatus)), emit: (body) => applyWith(() => body), peekRuntime: Ref.get(lifecycle).pipe(Effect.map((current) => current.held?.runtime)), - ensureRuntime, - releaseRuntime, + ensureRuntime: ensureRuntimeWithIdle, + releaseRuntime: cancelIdleTimer.pipe(Effect.andThen(releaseRuntime)), + suspendRuntime, } satisfies PiAgentSessionShape; }); diff --git a/packages/server/test/harness/event-manifest.test.ts b/packages/server/test/harness/event-manifest.test.ts index 09b643d59..f41ab92eb 100644 --- a/packages/server/test/harness/event-manifest.test.ts +++ b/packages/server/test/harness/event-manifest.test.ts @@ -44,6 +44,7 @@ describe("event manifest naming invariant", () => { expect(types).toContain("session.turn.ended"); expect(types).toContain("session.request.asked"); expect(types).toContain("session.queue.updated"); - expect(types).toContain("session.crashed"); + expect(types).toContain("session.crashed", + "session.runtime.stopped"); }); }); diff --git a/packages/server/test/harness/session.test.ts b/packages/server/test/harness/session.test.ts index 7d572feb3..6d4098f77 100644 --- a/packages/server/test/harness/session.test.ts +++ b/packages/server/test/harness/session.test.ts @@ -467,3 +467,65 @@ it.effect("a turn under the buffer caps is not marked truncated", () => }), ), ); + + +it.effect("suspendRuntime kills the process, leaves idle, and allows re-acquire", () => + run( + Effect.gen(function* () { + const session = yield* SessionService; + const queue = yield* makeQueue; + const closes = yield* Ref.make(0); + + yield* session.ensureRuntime(Effect.succeed(runtimeFrom(queue, { closes }))); + assert.ok(yield* session.peekRuntime); + const before = yield* session.snapshot; + + yield* session.suspendRuntime("idle"); + assert.equal(yield* Ref.get(closes), 1); + assert.equal(yield* session.peekRuntime, undefined); + const after = yield* session.snapshot; + assert.equal(after.status.phase, "idle"); + assert.ok(after.cursor > before.cursor); + + const replacement = yield* makeQueue; + const again = yield* session.ensureRuntime(Effect.succeed(runtimeFrom(replacement))); + assert.ok(again); + assert.ok(yield* session.peekRuntime); + }), + ), +); + +const FastIdleSessionLayer = Layer.effect( + SessionService, + Effect.gen(function* () { + const bus = yield* EventBus; + return yield* makePiAgentSession(ref, bus, { idleTimeoutMs: 40 }); + }), +).pipe(Layer.provide(EventBusLayer)); + +const runFastIdle = (program: Effect.Effect) => + program.pipe(Effect.provide(FastIdleSessionLayer)); + +it.effect("idle timeout suspends a held idle runtime", () => + runFastIdle( + Effect.gen(function* () { + const session = yield* SessionService; + const queue = yield* makeQueue; + const closes = yield* Ref.make(0); + yield* session.ensureRuntime(Effect.succeed(runtimeFrom(queue, { closes }))); + assert.ok(yield* session.peekRuntime); + + yield* Effect.sleep("100 millis"); + yield* Effect.eventually( + session.peekRuntime.pipe( + Effect.filterOrFail( + (runtime) => runtime === undefined, + () => new Error("runtime still held after idle timeout"), + ), + ), + ); + assert.equal(yield* Ref.get(closes), 1); + assert.equal((yield* session.status).phase, "idle"); + }), + ), +); diff --git a/packages/server/test/harness/types/session-envelope.test.ts b/packages/server/test/harness/types/session-envelope.test.ts index 62b677e00..22edbcf9a 100644 --- a/packages/server/test/harness/types/session-envelope.test.ts +++ b/packages/server/test/harness/types/session-envelope.test.ts @@ -10,6 +10,7 @@ describe("isSessionEvent", () => { "session.request.asked", "project.updated", "session.crashed", + "session.runtime.stopped", ]) { expect(isSessionEvent({ type } as never)).toBe(true); } From dd59dc766f6588c887e43669e0c63c319eadac0c Mon Sep 17 00:00:00 2001 From: owen Date: Sun, 6 Sep 2026 05:20:46 +0000 Subject: [PATCH 2/5] test: fix idle timeout clock and event-manifest assertions Use TestClock under @effect/vitest so the idle fiber actually fires, split toContain calls, and allow the stopped verb in the event naming invariant. --- packages/server/src/harness/session.ts | 5 ++++- .../server/test/harness/event-manifest.test.ts | 5 +++-- packages/server/test/harness/session.test.ts | 14 ++++---------- 3 files changed, 11 insertions(+), 13 deletions(-) diff --git a/packages/server/src/harness/session.ts b/packages/server/src/harness/session.ts index 509c049ec..900738422 100644 --- a/packages/server/src/harness/session.ts +++ b/packages/server/src/harness/session.ts @@ -458,7 +458,10 @@ export const makePiAgentSession = ( if (!held) return; const phase = yield* Ref.get(state).pipe(Effect.map((current) => current.phase)); if (phase !== "idle") return; - const generation = yield* Ref.updateAndGet(idleGeneration, (n) => n + 1); + const generation = yield* Ref.modify(idleGeneration, (n) => { + const next = n + 1; + return [next, next] as const; + }); yield* Effect.forkIn( identified( Effect.sleep(`${idleTimeoutMs} millis`).pipe( diff --git a/packages/server/test/harness/event-manifest.test.ts b/packages/server/test/harness/event-manifest.test.ts index f41ab92eb..b8b1b880d 100644 --- a/packages/server/test/harness/event-manifest.test.ts +++ b/packages/server/test/harness/event-manifest.test.ts @@ -13,6 +13,7 @@ const RESERVED_VERBS = new Set([ "replied", "rejected", "crashed", + "stopped", "failed", "exited", "connected", @@ -44,7 +45,7 @@ describe("event manifest naming invariant", () => { expect(types).toContain("session.turn.ended"); expect(types).toContain("session.request.asked"); expect(types).toContain("session.queue.updated"); - expect(types).toContain("session.crashed", - "session.runtime.stopped"); + expect(types).toContain("session.crashed"); + expect(types).toContain("session.runtime.stopped"); }); }); diff --git a/packages/server/test/harness/session.test.ts b/packages/server/test/harness/session.test.ts index 6d4098f77..45b98191d 100644 --- a/packages/server/test/harness/session.test.ts +++ b/packages/server/test/harness/session.test.ts @@ -4,6 +4,7 @@ import { it } from "@effect/vitest"; import type { AgentRequest, SessionRef } from "@getpie/contract"; import { Context, Effect, Layer, Queue, Ref, Stream } from "effect"; import type * as Cause from "effect/Cause"; +import { TestClock } from "effect/testing"; import { EventBus, EventBusLayer } from "../../src/events"; import { AgentOperationError, type SessionEnvelopeBody } from "../../src/harness"; @@ -468,7 +469,6 @@ it.effect("a turn under the buffer caps is not marked truncated", () => ), ); - it.effect("suspendRuntime kills the process, leaves idle, and allows re-acquire", () => run( Effect.gen(function* () { @@ -515,15 +515,9 @@ it.effect("idle timeout suspends a held idle runtime", () => yield* session.ensureRuntime(Effect.succeed(runtimeFrom(queue, { closes }))); assert.ok(yield* session.peekRuntime); - yield* Effect.sleep("100 millis"); - yield* Effect.eventually( - session.peekRuntime.pipe( - Effect.filterOrFail( - (runtime) => runtime === undefined, - () => new Error("runtime still held after idle timeout"), - ), - ), - ); + // @effect/vitest installs TestClock — real sleeps never fire the idle fiber. + yield* TestClock.adjust("100 millis"); + assert.equal(yield* session.peekRuntime, undefined); assert.equal(yield* Ref.get(closes), 1); assert.equal((yield* session.status).phase, "idle"); }), From ca23bba1c5c6adc29958b904f3a2d540550b14d9 Mon Sep 17 00:00:00 2001 From: dinq Date: Wed, 30 Sep 2026 13:29:02 -0400 Subject: [PATCH 3/5] style(server): format idle runtime configuration --- packages/server/src/harness/runtime-idle.ts | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/packages/server/src/harness/runtime-idle.ts b/packages/server/src/harness/runtime-idle.ts index 63f7ab728..90e5eea44 100644 --- a/packages/server/src/harness/runtime-idle.ts +++ b/packages/server/src/harness/runtime-idle.ts @@ -5,9 +5,7 @@ */ export const DEFAULT_SESSION_RUNTIME_IDLE_MS = 5 * 60 * 1000; -export const sessionRuntimeIdleMs = ( - env: NodeJS.ProcessEnv = process.env, -): number => { +export const sessionRuntimeIdleMs = (env: NodeJS.ProcessEnv = process.env): number => { const raw = env.PIE_SESSION_RUNTIME_IDLE_MS?.trim(); if (raw === undefined || raw === "") return DEFAULT_SESSION_RUNTIME_IDLE_MS; const parsed = Number(raw); From 81ceff43a1b4fc95dad90d1bd60859486ab3065e Mon Sep 17 00:00:00 2001 From: dinq Date: Wed, 30 Sep 2026 13:47:41 -0400 Subject: [PATCH 4/5] fix(server): serialize suspension with runtime reacquisition --- packages/server/src/harness/session.ts | 58 +++-- packages/server/test/harness/session.test.ts | 215 ++++++++++++++++++- 2 files changed, 253 insertions(+), 20 deletions(-) diff --git a/packages/server/src/harness/session.ts b/packages/server/src/harness/session.ts index 64b1e85c3..8cb0a94cf 100644 --- a/packages/server/src/harness/session.ts +++ b/packages/server/src/harness/session.ts @@ -65,8 +65,8 @@ type Held = { type Ticket = Deferred.Deferred; /** - * Whether this session has a runtime, is getting one, or will never take - * another — one Ref because the three have to be decided in a single step. + * Whether this session holds, acquires, or suspends a runtime, or will never + * take another — one Ref so ownership transitions happen in a single step. * Deliberately private: callers see `peekRuntime` / `ensureRuntime` / * `releaseRuntime` and nothing of the states in between. * @@ -80,6 +80,7 @@ type Ticket = Deferred.Deferred; type Lifecycle = { readonly held: Held | undefined; readonly acquiring: Ticket | undefined; + readonly suspending: Deferred.Deferred | undefined; readonly sealed: boolean; }; @@ -87,6 +88,7 @@ type AcquireDecision = | { readonly _tag: "Held"; readonly runtime: PiAgentRuntime } | { readonly _tag: "Start"; readonly ticket: Ticket } | { readonly _tag: "Await"; readonly ticket: Ticket } + | { readonly _tag: "Suspending"; readonly done: Deferred.Deferred } | { readonly _tag: "Sealed" }; export type PiAgentSessionShape = { @@ -152,6 +154,7 @@ export const makePiAgentSession = ( const lifecycle = yield* Ref.make({ held: undefined, acquiring: undefined, + suspending: undefined, sealed: false, }); // Generation bumps cancel an in-flight idle timer without racing suspend. @@ -362,6 +365,7 @@ export const makePiAgentSession = ( const ensureRuntime: PiAgentSessionShape["ensureRuntime"] = (acquire) => Ref.modify(lifecycle, (current): readonly [AcquireDecision, Lifecycle] => { if (current.sealed) return [{ _tag: "Sealed" }, current]; + if (current.suspending) return [{ _tag: "Suspending", done: current.suspending }, current]; if (current.held) return [{ _tag: "Held", runtime: current.held.runtime }, current]; if (current.acquiring) return [{ _tag: "Await", ticket: current.acquiring }, current]; const ticket = Deferred.makeUnsafe(); @@ -383,6 +387,8 @@ export const makePiAgentSession = ( switch (decision._tag) { case "Sealed": return Effect.succeed(undefined); + case "Suspending": + return Deferred.await(decision.done).pipe(Effect.andThen(ensureRuntime(acquire))); case "Held": return Effect.succeed(decision.runtime); case "Start": @@ -398,20 +404,21 @@ export const makePiAgentSession = ( const releaseRuntime: PiAgentSessionShape["releaseRuntime"] = Ref.modify( lifecycle, - (current) => - [current, { held: undefined, acquiring: current.acquiring, sealed: true }] as const, + (current) => [current, { ...current, held: undefined, sealed: true }] as const, ).pipe( Effect.flatMap((previous) => - previous.acquiring - ? // An acquisition already in flight owns a runtime it is about to - // store; wait for it and take that one, or the process it started - // would outlive the session that asked for it. - Deferred.await(previous.acquiring).pipe( - Effect.exit, - Effect.andThen(takeHeld), - Effect.flatMap(shutDown), - ) - : shutDown(previous.held), + previous.suspending + ? Deferred.await(previous.suspending) + : previous.acquiring + ? // An acquisition already in flight owns a runtime it is about to + // store; wait for it and take that one, or the process it started + // would outlive the session that asked for it. + Deferred.await(previous.acquiring).pipe( + Effect.exit, + Effect.andThen(takeHeld), + Effect.flatMap(shutDown), + ) + : shutDown(previous.held), ), Effect.uninterruptible, ); @@ -428,14 +435,21 @@ export const makePiAgentSession = ( Effect.andThen( Ref.modify(lifecycle, (current) => { if (current.sealed || !current.held) { - return [undefined as Held | undefined, current] as const; + return [undefined, current] as const; } - return [current.held, { ...current, held: undefined }] as const; + // Hide the closing runtime from peek, but retain ownership until + // shutdown and the stopped event finish. Neither acquisition nor + // release may overtake that event and reset a replacement turn. + const done = Deferred.makeUnsafe(); + return [ + { held: current.held, done }, + { ...current, held: undefined, suspending: done }, + ] as const; }), ), - Effect.flatMap((held) => - held - ? shutDown(held).pipe( + Effect.flatMap((pending) => + pending + ? shutDown(pending.held).pipe( Effect.andThen( apply({ type: "session.runtime.stopped", @@ -448,6 +462,12 @@ export const makePiAgentSession = ( Effect.annotateLogs({ event: "session.runtime.stopped", reason }), ), ), + Effect.ensuring( + Ref.update(lifecycle, (current) => ({ + ...current, + suspending: undefined, + })).pipe(Effect.andThen(Deferred.succeed(pending.done, undefined))), + ), ) : Effect.void, ), diff --git a/packages/server/test/harness/session.test.ts b/packages/server/test/harness/session.test.ts index 45b98191d..9e73fea28 100644 --- a/packages/server/test/harness/session.test.ts +++ b/packages/server/test/harness/session.test.ts @@ -2,7 +2,18 @@ import assert from "node:assert/strict"; import { it } from "@effect/vitest"; import type { AgentRequest, SessionRef } from "@getpie/contract"; -import { Context, Effect, Layer, Queue, Ref, Stream } from "effect"; +import { + Context, + Deferred, + Effect, + Exit, + Fiber, + Layer, + Queue, + Ref, + Scheduler, + Stream, +} from "effect"; import type * as Cause from "effect/Cause"; import { TestClock } from "effect/testing"; @@ -523,3 +534,205 @@ it.effect("idle timeout suspends a held idle runtime", () => }), ), ); + +// All descendant fibers share a dispatcher so flush reaches quiescence without +// wall-clock sleeps or an arbitrary number of scheduler yields. +const withControlledScheduler = ( + make: (flush: Effect.Effect) => Effect.Effect, +) => { + const dispatcher = new Scheduler.MixedScheduler().makeDispatcher(); + return make(Effect.sync(() => dispatcher.flush())).pipe( + Effect.provideService(Scheduler.Scheduler, { + executionMode: "async", + shouldYield: () => false, + makeDispatcher: () => dispatcher, + }), + ); +}; + +const holdSuspension = ( + session: PiAgentSessionShape, + afterClose: Effect.Effect = Effect.void, +) => + Effect.gen(function* () { + const queue = yield* makeQueue; + const closeStarted = yield* Deferred.make(); + const closeGate = yield* Deferred.make(); + const closes = yield* Ref.make(0); + yield* session.ensureRuntime( + Effect.succeed({ + ...runtimeFrom(queue), + close: Ref.update(closes, (n) => n + 1).pipe( + Effect.andThen(Deferred.succeed(closeStarted, undefined)), + Effect.andThen(Deferred.await(closeGate)), + Effect.andThen(afterClose), + ), + }), + ); + const suspending = yield* Effect.forkChild(session.suspendRuntime()); + yield* Deferred.await(closeStarted); + return { suspending, closeGate, closes }; + }); + +it.effect("suspendRuntime finishes stopping before a replacement turn starts", () => + withControlledScheduler((flush) => + run( + Effect.gen(function* () { + const session = yield* SessionService; + const { suspending, closeGate } = yield* holdSuspension(session); + + const queue = yield* makeQueue; + const replacement = runtimeFrom(queue); + const prompting = yield* Effect.forkChild( + session.ensureRuntime(Effect.succeed(replacement)).pipe( + Effect.flatMap((runtime) => { + assert.equal(runtime, replacement); + return Queue.offer(queue, { + type: "session.turn.started", + sessionId: nativeId, + turnId: "replacement-turn", + }); + }), + ), + ); + // Drain runnable fibers while close remains blocked: on the broken + // implementation the replacement turn is already running here. + yield* flush; + yield* Deferred.succeed(closeGate, undefined); + yield* Fiber.join(suspending); + yield* Fiber.join(prompting); + const snapshot = yield* awaitCursor(session, 2); + assert.equal(snapshot.status.phase, "running"); + assert.equal(snapshot.activeTurn?.turnId, "replacement-turn"); + }), + ), + ), +); + +it.effect("suspension preserves single-flight acquisition when a waiter is interrupted", () => + withControlledScheduler((flush) => + run( + Effect.gen(function* () { + const session = yield* SessionService; + const { suspending, closeGate, closes } = yield* holdSuspension(session); + const queue = yield* makeQueue; + const replacement = runtimeFrom(queue); + const acquisitions = yield* Ref.make(0); + const acquire = Ref.update(acquisitions, (n) => n + 1).pipe(Effect.as(replacement)); + const canceled = yield* Effect.forkChild(session.ensureRuntime(acquire)); + const first = yield* Effect.forkChild(session.ensureRuntime(acquire)); + const second = yield* Effect.forkChild(session.ensureRuntime(acquire)); + yield* flush; + yield* Fiber.interrupt(canceled); + const during = yield* Ref.get(acquisitions); + const peek = yield* session.peekRuntime; + // A second suspend must neither close twice nor invalidate ownership. + yield* session.suspendRuntime(); + const interruption = yield* Effect.forkChild(Fiber.interrupt(suspending)); + yield* flush; + yield* Deferred.succeed(closeGate, undefined); + yield* Fiber.join(interruption); + assert.equal(yield* Fiber.join(first), replacement); + assert.equal(yield* Fiber.join(second), replacement); + assert.equal(during, 0); + assert.equal(peek, undefined); + assert.equal(yield* Ref.get(acquisitions), 1); + assert.equal(yield* Ref.get(closes), 1); + assert.equal((yield* session.snapshot).cursor, 1); + }), + ), + ), +); + +it.effect("release waits for suspension and prevents a waiting acquisition", () => + withControlledScheduler((flush) => + run( + Effect.gen(function* () { + const session = yield* SessionService; + const { suspending, closeGate, closes } = yield* holdSuspension(session); + const queue = yield* makeQueue; + const acquisitions = yield* Ref.make(0); + const waiting = yield* Effect.forkChild( + session.ensureRuntime( + Ref.update(acquisitions, (n) => n + 1).pipe(Effect.as(runtimeFrom(queue))), + ), + ); + const released = yield* Ref.make(false); + const releasing = yield* Effect.forkChild( + session.releaseRuntime.pipe(Effect.andThen(Ref.set(released, true))), + ); + yield* flush; + const releasedEarly = yield* Ref.get(released); + yield* Deferred.succeed(closeGate, undefined); + yield* Fiber.join(suspending); + yield* Fiber.join(releasing); + assert.equal(yield* Fiber.join(waiting), undefined); + assert.equal(releasedEarly, false); + assert.equal(yield* Ref.get(acquisitions), 0); + assert.equal(yield* Ref.get(closes), 1); + assert.equal(yield* session.peekRuntime, undefined); + }), + ), + ), +); + +it.effect("stale idle timers cannot suspend a replacement running turn", () => + withControlledScheduler((flush) => + runFastIdle( + Effect.gen(function* () { + const session = yield* SessionService; + const { suspending, closeGate } = yield* holdSuspension(session); + yield* Deferred.succeed(closeGate, undefined); + yield* Fiber.join(suspending); + const queue = yield* makeQueue; + const closes = yield* Ref.make(0); + const replacement = runtimeFrom(queue, { closes }); + yield* session.ensureRuntime(Effect.succeed(replacement)); + yield* Queue.offer(queue, { + type: "session.turn.started", + sessionId: nativeId, + turnId: "replacement-turn", + }); + yield* flush; + yield* TestClock.adjust("100 millis"); + yield* flush; + assert.equal(yield* session.peekRuntime, replacement); + assert.equal(yield* Ref.get(closes), 0); + assert.equal((yield* session.status).phase, "running"); + yield* Queue.offer(queue, { + type: "session.turn.ended", + sessionId: nativeId, + turnId: "replacement-turn", + outcome: "completed", + }); + yield* flush; + yield* TestClock.adjust("100 millis"); + yield* flush; + assert.equal(yield* session.peekRuntime, undefined); + assert.equal(yield* Ref.get(closes), 1); + }), + ), + ), +); + +it.effect("a close defect does not strand acquisition behind suspension", () => + withControlledScheduler((flush) => + run( + Effect.gen(function* () { + const session = yield* SessionService; + const { suspending, closeGate } = yield* holdSuspension( + session, + Effect.die("close defect"), + ); + const queue = yield* makeQueue; + const replacement = runtimeFrom(queue); + const waiting = yield* Effect.forkChild(session.ensureRuntime(Effect.succeed(replacement))); + yield* flush; + yield* Deferred.succeed(closeGate, undefined); + assert.equal(Exit.isFailure(yield* Fiber.await(suspending)), true); + assert.equal(yield* Fiber.join(waiting), replacement); + assert.equal(yield* session.peekRuntime, replacement); + }), + ), + ), +); From 0cd6c976f34d1fd4e1058ce86609f9bcf97bf5c2 Mon Sep 17 00:00:00 2001 From: dinq Date: Wed, 30 Sep 2026 14:25:17 -0400 Subject: [PATCH 5/5] fix(server): reserve runtime during prompt admission --- .../server/src/harness/session-manager.ts | 9 + .../server/src/harness/session-service.ts | 3 +- packages/server/src/harness/session.ts | 160 +++++++++++----- .../test/harness/session-service.test.ts | 43 ++++- packages/server/test/harness/session.test.ts | 181 ++++++++++++++++++ 5 files changed, 341 insertions(+), 55 deletions(-) diff --git a/packages/server/src/harness/session-manager.ts b/packages/server/src/harness/session-manager.ts index e3a37bf9a..4419119f5 100644 --- a/packages/server/src/harness/session-manager.ts +++ b/packages/server/src/harness/session-manager.ts @@ -74,6 +74,8 @@ export type PiAgentSessionManagerShape = { * for reconnecting clients). */ readonly close: (ref: SessionRef) => Effect.Effect; + /** Reserve prompt admission before looking up a runtime; released with the caller scope. */ + readonly reserveAdmission: (ref: SessionRef) => Effect.Effect; /** * The status/snapshot of a session. Total on purpose: a ref with nothing * live in memory — the ordinary state of every persisted session after a @@ -222,6 +224,12 @@ export const makePiAgentSessionManager = ( Effect.flatMap((runtime) => (runtime ? Effect.succeed(runtime) : acquireVia(ref, acquire))), ); + const reserveAdmission: PiAgentSessionManagerShape["reserveAdmission"] = (ref) => + sessionFor(ref).pipe( + Effect.flatMap((session) => session.reserveAdmission), + Effect.flatMap((reserved) => (reserved ? Effect.void : reserveAdmission(ref))), + ); + const peek = (ref: SessionRef): Effect.Effect => withSession(ref, (session) => session.peekRuntime, undefined); @@ -292,6 +300,7 @@ export const makePiAgentSessionManager = ( ), ), peek, + reserveAdmission, close, status: (ref) => withSession(ref, (session) => session.status, toStatus(initialSessionState)), snapshot: (ref) => diff --git a/packages/server/src/harness/session-service.ts b/packages/server/src/harness/session-service.ts index 11de7177e..42dc04d39 100644 --- a/packages/server/src/harness/session-service.ts +++ b/packages/server/src/harness/session-service.ts @@ -459,9 +459,10 @@ export const PiAgentSessionServiceCoreLayer: Layer.Layer< > => Effect.gen(function* () { const resolved = yield* resolveWorkspace(ref); + yield* manager.reserveAdmission(ref); const runtime = yield* ensureRuntimeForPrompt(ref, resolved); return yield* runtime.prompt(userInput); - }); + }).pipe(Effect.scoped); const readHistory = ( ref: SessionRef, diff --git a/packages/server/src/harness/session.ts b/packages/server/src/harness/session.ts index 8cb0a94cf..099abd693 100644 --- a/packages/server/src/harness/session.ts +++ b/packages/server/src/harness/session.ts @@ -82,6 +82,8 @@ type Lifecycle = { readonly acquiring: Ticket | undefined; readonly suspending: Deferred.Deferred | undefined; readonly sealed: boolean; + readonly admissions: number; + readonly idleGeneration: number; }; type AcquireDecision = @@ -129,10 +131,13 @@ export type PiAgentSessionShape = { * rather than racing it. */ readonly releaseRuntime: Effect.Effect; + /** Protect prompt lookup/acquisition and admission from idle shutdown until scope exit. + * False means sealed; the manager must retry against its current session. */ + readonly reserveAdmission: Effect.Effect; /** * Kill the held Pi runtime without sealing the session, and publish * `session.runtime.stopped`. The next {@link ensureRuntime} may resume. - * No-op when nothing is held. Used for idle timeout. + * No-op unless held, idle, and free of prompt admissions. Used for idle timeout. */ readonly suspendRuntime: (reason?: string) => Effect.Effect; }; @@ -156,9 +161,9 @@ export const makePiAgentSession = ( acquiring: undefined, suspending: undefined, sealed: false, + admissions: 0, + idleGeneration: 0, }); - // Generation bumps cancel an in-flight idle timer without racing suspend. - const idleGeneration = yield* Ref.make(0); // Filled once idle helpers exist; applyWith calls it after every publish. let afterPublish: Effect.Effect = Effect.void; @@ -423,82 +428,130 @@ export const makePiAgentSession = ( Effect.uninterruptible, ); - const cancelIdleTimer: Effect.Effect = Ref.update(idleGeneration, (n) => n + 1); + const cancelIdleTimer: Effect.Effect = Ref.update(lifecycle, (current) => ({ + ...current, + idleGeneration: current.idleGeneration + 1, + })); /** * Drop the held runtime without sealing, publish `session.runtime.stopped`, * leave the session queryable so the next prompt can {@link ensureRuntime}. */ - const suspendRuntime = (reason = "idle"): Effect.Effect => + const suspendRuntime = (reason = "idle", generation?: number): Effect.Effect => identified( - cancelIdleTimer.pipe( - Effect.andThen( - Ref.modify(lifecycle, (current) => { - if (current.sealed || !current.held) { - return [undefined, current] as const; - } - // Hide the closing runtime from peek, but retain ownership until - // shutdown and the stopped event finish. Neither acquisition nor - // release may overtake that event and reset a replacement turn. - const done = Deferred.makeUnsafe(); - return [ - { held: current.held, done }, - { ...current, held: undefined, suspending: done }, - ] as const; + applyLock + .withPermit( + Effect.gen(function* () { + // Check eligibility and claim ownership under the fold/publication + // lock. Shutdown must run outside it: interrupting the drain and + // publishing stopped both need that lock to remain available. + if ((yield* Ref.get(state)).phase !== "idle") return undefined; + return yield* Ref.modify(lifecycle, (current) => { + if ( + current.sealed || + !current.held || + current.admissions > 0 || + (generation !== undefined && current.idleGeneration !== generation) + ) { + return [undefined, current] as const; + } + // Keep the ticket until both shutdown and stopped publication finish. + const done = Deferred.makeUnsafe(); + return [ + { held: current.held, done }, + { + ...current, + held: undefined, + suspending: done, + idleGeneration: current.idleGeneration + 1, + }, + ] as const; + }); }), - ), - Effect.flatMap((pending) => - pending - ? shutDown(pending.held).pipe( - Effect.andThen( - apply({ - type: "session.runtime.stopped", - sessionId: ref.sessionId, - reason, - }), - ), - Effect.andThen( - Effect.logInfo("session runtime suspended").pipe( - Effect.annotateLogs({ event: "session.runtime.stopped", reason }), + ) + .pipe( + Effect.flatMap((pending) => + pending + ? shutDown(pending.held).pipe( + Effect.andThen( + apply({ + type: "session.runtime.stopped", + sessionId: ref.sessionId, + reason, + }), + ), + Effect.andThen( + Effect.logInfo("session runtime suspended").pipe( + Effect.annotateLogs({ event: "session.runtime.stopped", reason }), + ), + ), + Effect.ensuring( + Ref.update(lifecycle, (current) => ({ + ...current, + suspending: undefined, + })).pipe(Effect.andThen(Deferred.succeed(pending.done, undefined))), ), - ), - Effect.ensuring( - Ref.update(lifecycle, (current) => ({ - ...current, - suspending: undefined, - })).pipe(Effect.andThen(Deferred.succeed(pending.done, undefined))), - ), - ) - : Effect.void, + ) + : Effect.void, + ), + Effect.uninterruptible, ), - Effect.uninterruptible, - ), ); /** Arm idle kill when we hold a runtime and the fold says idle. */ const armIdleTimer: Effect.Effect = Effect.gen(function* () { if (idleTimeoutMs <= 0) return; - const held = yield* Ref.get(lifecycle).pipe(Effect.map((current) => current.held)); - if (!held) return; + const ownership = yield* Ref.get(lifecycle); + if (!ownership.held || ownership.sealed || ownership.admissions > 0) return; const phase = yield* Ref.get(state).pipe(Effect.map((current) => current.phase)); if (phase !== "idle") return; - const generation = yield* Ref.modify(idleGeneration, (n) => { - const next = n + 1; - return [next, next] as const; + const generation = yield* Ref.modify(lifecycle, (current) => { + const next = current.idleGeneration + 1; + return [next, { ...current, idleGeneration: next }] as const; }); yield* Effect.forkIn( identified( Effect.sleep(`${idleTimeoutMs} millis`).pipe( - Effect.andThen(Ref.get(idleGeneration)), - Effect.flatMap((current) => - current === generation ? suspendRuntime("idle") : Effect.void, - ), + Effect.andThen(suspendRuntime("idle", generation)), ), ), ownerScope, ); }); + const reserveAdmission: PiAgentSessionShape["reserveAdmission"] = Effect.suspend(() => + Effect.acquireRelease( + Ref.modify( + lifecycle, + (current): readonly [boolean | Deferred.Deferred, Lifecycle] => { + if (current.sealed) return [false, current] as const; + if (current.suspending) return [current.suspending, current] as const; + return [ + true, + { + ...current, + admissions: current.admissions + 1, + idleGeneration: current.idleGeneration + 1, + }, + ] as const; + }, + ), + (reserved) => + reserved === true + ? Ref.update(lifecycle, (current) => ({ + ...current, + admissions: current.admissions - 1, + })).pipe(Effect.andThen(armIdleTimer)) + : Effect.void, + ).pipe( + Effect.flatMap((reserved) => + typeof reserved === "boolean" + ? Effect.succeed(reserved) + : Deferred.await(reserved).pipe(Effect.andThen(reserveAdmission)), + ), + ), + ); + // Wrap ensureRuntime to (re)arm idle once a runtime is held while idle. const ensureRuntimeWithIdle: PiAgentSessionShape["ensureRuntime"] = (acquire) => ensureRuntime(acquire).pipe(Effect.tap(() => armIdleTimer)); @@ -518,5 +571,6 @@ export const makePiAgentSession = ( ensureRuntime: ensureRuntimeWithIdle, releaseRuntime: cancelIdleTimer.pipe(Effect.andThen(releaseRuntime)), suspendRuntime, + reserveAdmission, } satisfies PiAgentSessionShape; }); diff --git a/packages/server/test/harness/session-service.test.ts b/packages/server/test/harness/session-service.test.ts index aa99241d2..6e6f0d75a 100644 --- a/packages/server/test/harness/session-service.test.ts +++ b/packages/server/test/harness/session-service.test.ts @@ -2,7 +2,8 @@ import assert from "node:assert/strict"; import { layer } from "@effect/vitest"; import { isSessionScopedEvent, type SessionRef, type PieUIMessage } from "@getpie/contract"; -import { Effect, Fiber, FileSystem, Layer, Logger, References, Stream } from "effect"; +import { Deferred, Effect, Fiber, FileSystem, Layer, Logger, References, Stream } from "effect"; +import { TestClock } from "effect/testing"; import { structured, type LogRecord } from "../log-record"; import { NodePlatformLayer } from "../platform"; @@ -33,6 +34,46 @@ layer(NodePlatformLayer)("PiAgentSessionService", (it) => { }), ); + it.effect("idle deadline does not close a runtime during prompt admission", () => + Effect.gen(function* () { + const admitting = yield* Deferred.make(); + const response = yield* Deferred.make(); + let calls = 0; + yield* run( + { + turn: "finished", + prompt: () => + Effect.gen(function* () { + calls += 1; + if (calls > 1) { + yield* Deferred.succeed(admitting, undefined); + yield* Deferred.await(response); + } + return { turnId: "turn-1", started: true }; + }), + }, + (fixture) => + Effect.gen(function* () { + const { ref } = yield* createSession(fixture); + const input = { ref, parts: [{ type: "text" as const, text: "hello" }] }; + yield* fixture.service.prompt(input); + yield* TestClock.adjust("299 seconds"); + const pending = yield* Effect.forkChild(fixture.service.prompt(input)); + yield* Deferred.await(admitting); + yield* TestClock.adjust("301 seconds"); + const closedDuringAdmission = fixture.spy.close.slice(); + yield* Deferred.succeed(response, undefined); + assert.equal((yield* Fiber.join(pending)).started, true); + assert.deepEqual(closedDuringAdmission, []); + assert.equal(fixture.spy.open.length, 1); + assert.equal(fixture.spy.resume.length, 0); + yield* TestClock.adjust("301 seconds"); + assert.deepEqual(fixture.spy.close, ["native-1"]); + }), + ); + }), + ); + it.effect("create persists an optional title", () => Effect.gen(function* () { const result = yield* run({}, (fixture) => diff --git a/packages/server/test/harness/session.test.ts b/packages/server/test/harness/session.test.ts index 9e73fea28..3aae6bec1 100644 --- a/packages/server/test/harness/session.test.ts +++ b/packages/server/test/harness/session.test.ts @@ -736,3 +736,184 @@ it.effect("a close defect does not strand acquisition behind suspension", () => ), ), ); + +it.effect("idle deadline cannot interrupt turn-start publication", () => + withControlledScheduler((flush) => + Effect.scoped( + Effect.gen(function* () { + const bus = yield* Effect.provide(EventBus, EventBusLayer); + const publishing = yield* Deferred.make(); + const publishGate = yield* Deferred.make(); + const session = yield* makePiAgentSession( + ref, + { + ...bus, + publish: (event) => + event.type === "session.turn.started" + ? Deferred.succeed(publishing, undefined).pipe( + Effect.andThen(Deferred.await(publishGate)), + Effect.andThen(bus.publish(event)), + ) + : bus.publish(event), + }, + { idleTimeoutMs: 40 }, + ); + const queue = yield* makeQueue; + const closes = yield* Ref.make(0); + const runtime = runtimeFrom(queue, { closes }); + yield* session.ensureRuntime(Effect.succeed(runtime)); + yield* Queue.offer(queue, { + type: "session.turn.started", + sessionId: nativeId, + turnId: "turn-1", + }); + yield* Deferred.await(publishing); + assert.equal((yield* session.status).phase, "running"); + yield* TestClock.adjust("100 millis"); + yield* flush; + const heldAtDeadline = yield* session.peekRuntime; + const closesAtDeadline = yield* Ref.get(closes); + yield* Deferred.succeed(publishGate, undefined); + yield* flush; + assert.equal(heldAtDeadline, runtime); + assert.equal(closesAtDeadline, 0); + assert.equal((yield* session.snapshot).status.phase, "running"); + }), + ), + ), +); + +for (const ending of ["success", "failure", "interruption"] as const) { + it.effect(`admission reservation releases after ${ending} without releasing another caller`, () => + withControlledScheduler((flush) => + Effect.scoped( + Effect.gen(function* () { + const bus = yield* Effect.provide(EventBus, EventBusLayer); + const session = yield* makePiAgentSession(ref, bus, { idleTimeoutMs: 40 }); + const queue = yield* makeQueue; + const closes = yield* Ref.make(0); + const runtime = runtimeFrom(queue, { closes }); + yield* session.ensureRuntime(Effect.succeed(runtime)); + const firstGate = yield* Deferred.make(); + const secondGate = yield* Deferred.make(); + const first = yield* Effect.forkChild( + Effect.scoped( + session.reserveAdmission.pipe( + Effect.andThen(Deferred.await(firstGate)), + Effect.andThen(ending === "failure" ? Effect.fail("rejected") : Effect.void), + ), + ), + ); + const second = yield* Effect.forkChild( + Effect.scoped( + session.reserveAdmission.pipe(Effect.andThen(Deferred.await(secondGate))), + ), + ); + yield* flush; + yield* TestClock.adjust("100 millis"); + assert.equal(yield* Ref.get(closes), 0); + if (ending === "interruption") yield* Fiber.interrupt(first); + else { + yield* Deferred.succeed(firstGate, undefined); + assert.equal(Exit.isFailure(yield* Fiber.await(first)), ending === "failure"); + } + yield* TestClock.adjust("100 millis"); + assert.equal(yield* session.peekRuntime, runtime); + yield* Deferred.succeed(secondGate, undefined); + yield* Fiber.join(second); + yield* TestClock.adjust("39 millis"); + assert.equal(yield* Ref.get(closes), 0); + yield* TestClock.adjust("1 millis"); + yield* flush; + assert.equal(yield* Ref.get(closes), 1); + yield* session.releaseRuntime; + assert.equal(yield* session.reserveAdmission, false); + }), + ), + ), + ); +} + +it.effect("admission waits through stopped publication and an interrupted waiter", () => + withControlledScheduler((flush) => + Effect.scoped( + Effect.gen(function* () { + const bus = yield* Effect.provide(EventBus, EventBusLayer); + const publishing = yield* Deferred.make(); + const publishGate = yield* Deferred.make(); + const session = yield* makePiAgentSession( + ref, + { + ...bus, + publish: (event) => + event.type === "session.runtime.stopped" + ? Deferred.succeed(publishing, undefined).pipe( + Effect.andThen(Deferred.await(publishGate)), + Effect.andThen(bus.publish(event)), + ) + : bus.publish(event), + }, + { idleTimeoutMs: 40 }, + ); + const { suspending, closeGate } = yield* holdSuspension(session); + let admitted = false; + const canceled = yield* Effect.forkChild(Effect.scoped(session.reserveAdmission)); + const waiting = yield* Effect.forkChild( + Effect.scoped( + session.reserveAdmission.pipe( + Effect.tap(() => + Effect.sync(() => { + admitted = true; + }), + ), + ), + ), + ); + yield* flush; + assert.equal(admitted, false); + yield* Fiber.interrupt(canceled); + yield* Deferred.succeed(closeGate, undefined); + yield* Deferred.await(publishing); + yield* flush; + assert.equal(admitted, false); + yield* Deferred.succeed(publishGate, undefined); + yield* Fiber.join(suspending); + assert.equal(yield* Fiber.join(waiting), true); + assert.equal(admitted, true); + const queue = yield* makeQueue; + const closes = yield* Ref.make(0); + yield* session.ensureRuntime(Effect.succeed(runtimeFrom(queue, { closes }))); + yield* TestClock.adjust("100 millis"); + yield* flush; + assert.equal(yield* Ref.get(closes), 1); + }), + ), + ), +); + +for (const phase of ["running", "requires_action", "disabled"] as const) { + it.effect(`idle timeout preserves a ${phase} runtime`, () => + Effect.scoped( + Effect.gen(function* () { + const bus = yield* Effect.provide(EventBus, EventBusLayer); + const session = yield* makePiAgentSession(ref, bus, { + idleTimeoutMs: phase === "disabled" ? 0 : 40, + }); + const queue = yield* makeQueue; + const closes = yield* Ref.make(0); + const runtime = runtimeFrom(queue, { closes }); + yield* session.ensureRuntime(Effect.succeed(runtime)); + if (phase !== "disabled") + yield* session.emit({ type: "session.turn.started", turnId: "turn-1" }); + if (phase === "requires_action") + yield* session.emit({ + type: "session.request.asked", + request: { type: "question", id: "request-1", questions: [], native: null }, + }); + yield* TestClock.adjust("100 millis"); + assert.equal(yield* session.peekRuntime, runtime); + assert.equal(yield* Ref.get(closes), 0); + }), + ), + ); +}