diff --git a/apps/server/src/persistence/ProviderSessionRuntime.ts b/apps/server/src/persistence/ProviderSessionRuntime.ts index a3475d2f190..95bd1ee6d3f 100644 --- a/apps/server/src/persistence/ProviderSessionRuntime.ts +++ b/apps/server/src/persistence/ProviderSessionRuntime.ts @@ -56,6 +56,12 @@ export type GetProviderSessionRuntimeInput = typeof GetProviderSessionRuntimeInp export const DeleteProviderSessionRuntimeInput = Schema.Struct({ threadId: ThreadId }); export type DeleteProviderSessionRuntimeInput = typeof DeleteProviderSessionRuntimeInput.Type; +export const TouchProviderSessionRuntimeInput = Schema.Struct({ + threadId: ThreadId, + lastSeenAt: IsoDateTime, +}); +export type TouchProviderSessionRuntimeInput = typeof TouchProviderSessionRuntimeInput.Type; + /** * ProviderSessionRuntimeRepository - Service tag for provider runtime persistence. */ @@ -97,6 +103,16 @@ export class ProviderSessionRuntimeRepository extends Context.Service< readonly deleteByThreadId: ( input: DeleteProviderSessionRuntimeInput, ) => Effect.Effect; + + /** + * Bump `last_seen_at` for an existing runtime row. + * + * No-op when no row matches. Cheaper than `upsert` for high-frequency + * activity signals because it skips the SELECT and JSON re-encoding. + */ + readonly touchByThreadId: ( + input: TouchProviderSessionRuntimeInput, + ) => Effect.Effect; } >()("t3/persistence/ProviderSessionRuntime/ProviderSessionRuntimeRepository") {} @@ -233,6 +249,16 @@ export const make = Effect.gen(function* () { `, }); + const touchRuntimeByThreadId = SqlSchema.void({ + Request: TouchProviderSessionRuntimeInput, + execute: ({ threadId, lastSeenAt }) => + sql` + UPDATE provider_session_runtime + SET last_seen_at = ${lastSeenAt} + WHERE thread_id = ${threadId} + `, + }); + const upsert: ProviderSessionRuntimeRepository["Service"]["upsert"] = (runtime) => upsertRuntimeRow(runtime).pipe( Effect.mapError( @@ -308,11 +334,23 @@ export const make = Effect.gen(function* () { ), ); + const touchByThreadId: ProviderSessionRuntimeRepository["Service"]["touchByThreadId"] = (input) => + touchRuntimeByThreadId(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProviderSessionRuntimeRepository.touchByThreadId:query", + "ProviderSessionRuntimeRepository.touchByThreadId:encodeRequest", + { threadId: input.threadId }, + ), + ), + ); + return { upsert, getByThreadId, list, deleteByThreadId, + touchByThreadId, } satisfies ProviderSessionRuntimeRepository["Service"]; }); diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 515a7c6fcbb..2e4f9b97149 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -213,6 +213,7 @@ function makeScopedRuntimeFactory(options?: { readonly failConstruction?: boolea const providerSessionDirectoryTestLayer = Layer.succeed(ProviderSessionDirectory, { upsert: () => Effect.void, + touch: () => Effect.void, getProvider: () => Effect.die(new Error("ProviderSessionDirectory.getProvider is not used in test")), getBinding: () => Effect.succeed(Option.none()), diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index d0475e25284..6ea17281b5e 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -180,6 +180,7 @@ const OpenCodeRuntimeTestDouble: OpenCodeRuntimeShape = { const providerSessionDirectoryTestLayer = Layer.succeed(ProviderSessionDirectory, { upsert: () => Effect.void, + touch: () => Effect.void, getProvider: () => Effect.die(new Error("ProviderSessionDirectory.getProvider is not used in test")), getBinding: () => Effect.succeed(Option.none()), diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 2eaaeb8ce3c..7983752fdf8 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -25,6 +25,7 @@ import { type ProviderSession, } from "@t3tools/contracts"; import { causeErrorTag } from "@t3tools/shared/observability"; +import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; @@ -281,6 +282,30 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( }); }); + // Runtime event types that must NOT count as session activity. + // `session.exited` means the provider process is gone — bumping `lastSeenAt` + // would keep the directory binding alive past the runtime that backs it, + // defeating the reaper. Extend this set if new event types turn out to fire + // after a session is effectively dead. + const SESSION_ACTIVITY_DENY_LIST: ReadonlySet = new Set([ + "session.exited", + ]); + + const bumpSessionActivity = (event: ProviderRuntimeEvent): Effect.Effect => + SESSION_ACTIVITY_DENY_LIST.has(event.type) + ? Effect.void + : directory.touch(event.threadId).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : Effect.logDebug("provider.session.activity.touch-failed", { + threadId: event.threadId, + eventType: event.type, + cause, + }), + ), + ); + const processRuntimeEvent = ( source: { readonly instanceId: ProviderInstanceId; @@ -293,7 +318,10 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( increment(providerRuntimeEventsTotal, { provider: canonicalEvent.provider, eventType: canonicalEvent.type, - }).pipe(Effect.andThen(publishRuntimeEvent(canonicalEvent))), + }).pipe( + Effect.andThen(publishRuntimeEvent(canonicalEvent)), + Effect.andThen(bumpSessionActivity(canonicalEvent)), + ), ), ); diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts index 23075bd9a06..cbce06db5da 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts @@ -148,6 +148,13 @@ const makeProviderSessionDirectory = Effect.gen(function* () { .pipe(Effect.mapError(toPersistenceError("ProviderSessionDirectory.upsert:upsert"))); }); + const touch: ProviderSessionDirectoryShape["touch"] = Effect.fn(function* (threadId) { + const now = DateTime.formatIso(yield* DateTime.now); + yield* repository + .touchByThreadId({ threadId, lastSeenAt: now }) + .pipe(Effect.mapError(toPersistenceError("ProviderSessionDirectory.touch:touchByThreadId"))); + }); + const getProvider: ProviderSessionDirectoryShape["getProvider"] = (threadId) => getBinding(threadId).pipe( Effect.flatMap((binding) => @@ -184,6 +191,7 @@ const makeProviderSessionDirectory = Effect.gen(function* () { return { upsert, + touch, getProvider, getBinding, listThreadIds, diff --git a/apps/server/src/provider/Services/ProviderSessionDirectory.ts b/apps/server/src/provider/Services/ProviderSessionDirectory.ts index f2dd4323f7a..c27a82bba0b 100644 --- a/apps/server/src/provider/Services/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Services/ProviderSessionDirectory.ts @@ -45,6 +45,10 @@ export interface ProviderSessionDirectoryShape { binding: ProviderRuntimeBinding, ) => Effect.Effect; + readonly touch: ( + threadId: ThreadId, + ) => Effect.Effect; + readonly getProvider: ( threadId: ThreadId, ) => Effect.Effect;