diff --git a/CONTEXT.md b/CONTEXT.md index 0ca8c248f..ffd6167cc 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -58,6 +58,9 @@ Always Bun: `bun --mode rpc …`. `@getpie/server#build` emits t Always Node. Desktop spawns Electron-as-Node (`Pie Helper` + asar `server.mjs`, `ELECTRON_RUN_AS_NODE`). CLI uses `process.execPath`. The live terminal is `node-pty`. Bun is only for pie-pi-process (PATH `bun` plus the package export). _Avoid_: spawning the shebang `pi` binary under Bun; using a user-installed `pi` as `pie-pi-process`; a Node spawn path for pie-pi-process; a Bun runtime for the daemon or live terminal +**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 efc40b577..abd3be42d 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 0e20a1aff..019688df9 100644 --- a/apps/app/src/features/chat/runtime/chat.ts +++ b/apps/app/src/features/chat/runtime/chat.ts @@ -238,6 +238,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 2a8cf11d9..247350b3f 100644 --- a/packages/contract/src/domain.ts +++ b/packages/contract/src/domain.ts @@ -197,6 +197,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]; @@ -255,7 +258,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 b4f072084..2e06afa62 100644 --- a/packages/server/src/harness/events/framework.ts +++ b/packages/server/src/harness/events/framework.ts @@ -78,6 +78,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 }, @@ -107,6 +111,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..90e5eea44 --- /dev/null +++ b/packages/server/src/harness/runtime-idle.ts @@ -0,0 +1,14 @@ +/** + * 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 31ebeefce..9881beee9 100644 --- a/packages/server/src/harness/session-fold.ts +++ b/packages/server/src/harness/session-fold.ts @@ -135,6 +135,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 }; default: { const exhaustive: never = event; return exhaustive; @@ -261,6 +263,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, + }; default: { const exhaustive: never = event; return exhaustive; 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 eed19f852..099abd693 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, @@ -64,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. * @@ -79,13 +80,17 @@ type Ticket = Deferred.Deferred; type Lifecycle = { readonly held: Held | undefined; readonly acquiring: Ticket | undefined; + readonly suspending: Deferred.Deferred | undefined; readonly sealed: boolean; + readonly admissions: number; + readonly idleGeneration: number; }; 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 = { @@ -126,20 +131,41 @@ 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 unless held, idle, and free of prompt admissions. 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, + suspending: undefined, sealed: false, + admissions: 0, + idleGeneration: 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 +241,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)), ); }), ), @@ -343,6 +370,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(); @@ -364,6 +392,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": @@ -379,24 +409,157 @@ 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, ); + 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", generation?: number): Effect.Effect => + identified( + 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; + }); + }), + ) + .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.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 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(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(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)); + + afterPublish = Ref.get(state).pipe( + Effect.flatMap((current) => (current.phase === "idle" ? armIdleTimer : cancelIdleTimer)), + ); + return { ref, snapshot: applyLock.withPermit( @@ -405,7 +568,9 @@ 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, + reserveAdmission, } satisfies PiAgentSessionShape; }); diff --git a/packages/server/test/harness/event-manifest.test.ts b/packages/server/test/harness/event-manifest.test.ts index f3b7d6b7a..85db2595d 100644 --- a/packages/server/test/harness/event-manifest.test.ts +++ b/packages/server/test/harness/event-manifest.test.ts @@ -14,6 +14,7 @@ const RESERVED_VERBS = new Set([ "replied", "rejected", "crashed", + "stopped", "failed", "exited", "connected", @@ -47,5 +48,6 @@ describe("event manifest naming invariant", () => { expect(types).toContain("session.request.asked"); expect(types).toContain("session.queue.updated"); expect(types).toContain("session.crashed"); + expect(types).toContain("session.runtime.stopped"); }); }); 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 7d572feb3..3aae6bec1 100644 --- a/packages/server/test/harness/session.test.ts +++ b/packages/server/test/harness/session.test.ts @@ -2,8 +2,20 @@ 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"; import { EventBus, EventBusLayer } from "../../src/events"; import { AgentOperationError, type SessionEnvelopeBody } from "../../src/harness"; @@ -467,3 +479,441 @@ 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); + + // @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"); + }), + ), +); + +// 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); + }), + ), + ), +); + +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); + }), + ), + ); +} 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); }