From 9e5c4442cf4abfa4573aed04156a2481b9580cc8 Mon Sep 17 00:00:00 2001 From: ozppupbg <43532395+ozppupbg@users.noreply.github.com> Date: Fri, 10 Jul 2026 15:41:56 +0200 Subject: [PATCH 1/2] fix(server): treat provider runtime output as session activity Change activity tracking to cover provider side, not only user actions. This keeps the provider session active while (sub)agents are running. Partially addresses #2980 & #2378. Co-authored-by: Claude Opus 4.7 --- .../src/persistence/ProviderSessionRuntime.ts | 38 +++++++++++++++++++ .../src/provider/Layers/CodexAdapter.test.ts | 1 + .../provider/Layers/OpenCodeAdapter.test.ts | 1 + .../src/provider/Layers/ProviderService.ts | 27 ++++++++++++- .../Layers/ProviderSessionDirectory.ts | 8 ++++ .../Services/ProviderSessionDirectory.ts | 4 ++ 6 files changed, 78 insertions(+), 1 deletion(-) 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..e670fbf63e0 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -281,6 +281,28 @@ 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) => + Effect.logDebug("provider.session.activity.touch-failed", { + threadId: event.threadId, + eventType: event.type, + cause, + }), + ), + ); + const processRuntimeEvent = ( source: { readonly instanceId: ProviderInstanceId; @@ -293,7 +315,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; From bd8a0a9b4a3c9a6d321596821b35008672b9c4d0 Mon Sep 17 00:00:00 2001 From: ozppupbg <43532395+ozppupbg@users.noreply.github.com> Date: Fri, 10 Jul 2026 17:43:14 +0200 Subject: [PATCH 2/2] Re-raise interrupt-only causes so a shutting-down subscription fiber still unwinds; only a genuine touch failure is swallowed. Co-authored-by: Claude Opus 4.8 --- apps/server/src/provider/Layers/ProviderService.ts | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index e670fbf63e0..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"; @@ -295,11 +296,13 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ? Effect.void : directory.touch(event.threadId).pipe( Effect.catchCause((cause) => - Effect.logDebug("provider.session.activity.touch-failed", { - threadId: event.threadId, - eventType: event.type, - cause, - }), + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : Effect.logDebug("provider.session.activity.touch-failed", { + threadId: event.threadId, + eventType: event.type, + cause, + }), ), );