diff --git a/apps/server/src/persistence/ProviderSessionRuntime.ts b/apps/server/src/persistence/ProviderSessionRuntime.ts index 2ccdd862522..a778e4bed03 100644 --- a/apps/server/src/persistence/ProviderSessionRuntime.ts +++ b/apps/server/src/persistence/ProviderSessionRuntime.ts @@ -58,6 +58,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. */ @@ -73,6 +79,15 @@ export class ProviderSessionRuntimeRepository extends Context.Service< runtime: ProviderSessionRuntime, ) => Effect.Effect; + /** + * Update only `last_seen_at` for an existing runtime row. + * + * No-op when no row exists for the thread; never creates a row. + */ + readonly touchByThreadId: ( + input: TouchProviderSessionRuntimeInput, + ) => Effect.Effect; + /** * Read provider runtime state by canonical thread id. */ @@ -186,6 +201,16 @@ export const make = Effect.gen(function* () { `, }); + const touchRuntimeRow = SqlSchema.void({ + Request: TouchProviderSessionRuntimeInput, + execute: ({ threadId, lastSeenAt }) => + sql` + UPDATE provider_session_runtime + SET last_seen_at = ${lastSeenAt} + WHERE thread_id = ${threadId} + `, + }); + const getRuntimeRowByThreadId = SqlSchema.findOneOption({ Request: GetRuntimeRequestSchema, Result: ProviderSessionRuntimeRawDbRowSchema, @@ -246,6 +271,17 @@ export const make = Effect.gen(function* () { ), ); + const touchByThreadId: ProviderSessionRuntimeRepository["Service"]["touchByThreadId"] = (input) => + touchRuntimeRow(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProviderSessionRuntimeRepository.touchByThreadId:query", + "ProviderSessionRuntimeRepository.touchByThreadId:encodeRequest", + { threadId: input.threadId }, + ), + ), + ); + const getByThreadId: ProviderSessionRuntimeRepository["Service"]["getByThreadId"] = (input) => getRuntimeRowByThreadId(input).pipe( Effect.mapError( @@ -324,6 +360,7 @@ export const make = Effect.gen(function* () { return { upsert, + touchByThreadId, getByThreadId, list, deleteByThreadId, diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 7b8fbec5666..f51a3e2c6ee 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 1385ccbaabe..31fe8328cbd 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -233,6 +233,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.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index ccbbce1759f..8392a52f458 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -46,7 +46,7 @@ import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts"; import * as ProviderAdapterRegistry from "../Services/ProviderAdapterRegistry.ts"; import * as ProviderService from "../Services/ProviderService.ts"; import * as ProviderSessionDirectory from "../Services/ProviderSessionDirectory.ts"; -import { makeProviderServiceLive } from "./ProviderService.ts"; +import { makeProviderServiceLive, SESSION_ACTIVITY_EVENT_TYPES } from "./ProviderService.ts"; import * as ProviderEventLoggers from "./ProviderEventLoggers.ts"; import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts"; import * as NodeServices from "@effect/platform-node/NodeServices"; @@ -1536,6 +1536,68 @@ fanout.layer("ProviderServiceLive fanout", (it) => { }), ); + it.effect("refreshes binding lastSeenAt on turn and task lifecycle events", () => + Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; + const threadId = asThreadId("thread-last-seen"); + yield* provider.startSession(threadId, { + provider: ProviderDriverKind.make("codex"), + providerInstanceId: codexInstanceId, + threadId, + runtimeMode: "full-access", + }); + + const lastSeenAt = directory + .listBindings() + .pipe( + Effect.map( + (bindings) => bindings.find((entry) => entry.threadId === threadId)?.lastSeenAt, + ), + ); + + const emitEvent = (type: string, eventId: string) => { + fanout.codex.emit({ + type, + eventId: asEventId(eventId), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + threadId, + turnId: asTurnId("turn-last-seen"), + }); + }; + + const initial = yield* lastSeenAt; + assert.notEqual(initial, undefined); + yield* advanceTestClock(50); + + // A content delta is not a session-activity boundary and must not + // write to the binding on every streamed token. + emitEvent("content.delta", "evt-seen-delta"); + yield* advanceTestClock(50); + const afterDelta = yield* lastSeenAt; + assert.equal(afterDelta, initial); + + let previous = initial as string; + for (const [index, type] of [...SESSION_ACTIVITY_EVENT_TYPES].entries()) { + yield* advanceTestClock(61_000); + emitEvent(type, `evt-seen-${index}`); + yield* advanceTestClock(50); + const current = yield* lastSeenAt; + assert.notEqual(current, undefined); + assert.equal(Date.parse(current as string) > Date.parse(previous), true); + previous = current as string; + } + + // Touches are throttled per thread: a boundary arriving right after + // another one must not hit the database again. + emitEvent("turn.completed", "evt-seen-throttled"); + yield* advanceTestClock(50); + const afterThrottled = yield* lastSeenAt; + assert.equal(afterThrottled, previous); + }), + ); + it.effect("fans out canonical runtime events in emission order", () => Effect.gen(function* () { const provider = yield* ProviderService.ProviderService; diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index ecf26a914c1..b530e9a916c 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 Clock from "effect/Clock"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; @@ -179,6 +180,18 @@ const dieOnMissingBindingInstanceId = ( ); }; +export const SESSION_ACTIVITY_EVENT_TYPES: ReadonlySet = new Set([ + "turn.started", + "turn.completed", + "turn.aborted", + "task.started", + "task.progress", + "task.updated", + "task.completed", +]); + +const SESSION_ACTIVITY_TOUCH_MIN_INTERVAL_MS = 60_000; + const correlateRuntimeEventWithInstance = ( source: { readonly instanceId: ProviderInstanceId; @@ -281,6 +294,45 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( }); }); + // Turn and background-task lifecycle events prove the provider session is + // still doing user-visible work even when no `sendTurn` call delivered it + // (queued follow-ups, plan-sourced turns, background tasks that outlive + // their foreground turn). Without this, `lastSeenAt` stays at the last + // explicit send and the inactivity reaper measures idleness from the last + // user message instead of the last completed work. + // + // Touches are throttled per thread because this runs on the event drain + // fiber and task.progress can fire many times per second in multi-agent + // sessions; the reaper thresholds are minutes, so minute-level lastSeenAt + // granularity is enough. The gate timestamp is set before the write so a + // failing database is not retried on every event. + const lastTouchAtByThread = new Map(); + + const touchSessionActivity = (event: ProviderRuntimeEvent): Effect.Effect => + SESSION_ACTIVITY_EVENT_TYPES.has(event.type) + ? Clock.currentTimeMillis.pipe( + Effect.flatMap((now) => { + const lastTouchAt = lastTouchAtByThread.get(event.threadId); + if ( + lastTouchAt !== undefined && + now - lastTouchAt < SESSION_ACTIVITY_TOUCH_MIN_INTERVAL_MS + ) { + return Effect.void; + } + lastTouchAtByThread.set(event.threadId, now); + return directory.touch(event.threadId).pipe( + Effect.catchCause((cause) => + Effect.logWarning("provider.session.activity-touch-failed", { + threadId: event.threadId, + eventType: event.type, + errorTag: causeErrorTag(cause), + }), + ), + ); + }), + ) + : Effect.void; + const processRuntimeEvent = ( source: { readonly instanceId: ProviderInstanceId; @@ -293,7 +345,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(touchSessionActivity(canonicalEvent)), + ), ), ); @@ -856,6 +911,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( yield* routed.adapter.stopSession(routed.threadId); } yield* clearMcpSession(input.threadId); + lastTouchAtByThread.delete(input.threadId); yield* directory.upsert({ threadId: input.threadId, provider: routed.adapter.provider, diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts index 079b7f10ebf..43c239d373d 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts @@ -227,6 +227,61 @@ it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryL } })); + it.effect("touch refreshes lastSeenAt and leaves every other field unchanged", () => + Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory; + const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + const threadId = ThreadId.make("thread-touch"); + + yield* runtimeRepository.upsert({ + threadId, + providerName: "claudeAgent", + providerInstanceId: null, + adapterKey: "claudeAgent", + runtimeMode: "full-access", + status: "running", + lastSeenAt: "2026-01-01T00:00:00.000Z", + resumeCursor: { + opaque: "resume-touch", + }, + runtimePayload: { + cwd: "/tmp/touch", + }, + }); + + yield* directory.touch(threadId); + + const runtime = yield* runtimeRepository.getByThreadId({ threadId }); + assert.equal(Option.isSome(runtime), true); + if (Option.isSome(runtime)) { + assert.notEqual(runtime.value.lastSeenAt, "2026-01-01T00:00:00.000Z"); + assert.equal(Number.isNaN(Date.parse(runtime.value.lastSeenAt)), false); + assert.equal(runtime.value.providerName, "claudeAgent"); + assert.equal(runtime.value.adapterKey, "claudeAgent"); + assert.equal(runtime.value.runtimeMode, "full-access"); + assert.equal(runtime.value.status, "running"); + assert.deepEqual(runtime.value.resumeCursor, { + opaque: "resume-touch", + }); + assert.deepEqual(runtime.value.runtimePayload, { + cwd: "/tmp/touch", + }); + } + }), + ); + + it.effect("touch never creates a row for an unknown thread", () => + Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory; + const threadId = ThreadId.make("thread-touch-missing"); + + yield* directory.touch(threadId); + + const binding = yield* directory.getBinding(threadId); + assert.equal(Option.isNone(binding), true); + }), + ); + it("rehydrates persisted mappings across layer restart", () => Effect.gen(function* () { const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-directory-")); 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..898739a9105 100644 --- a/apps/server/src/provider/Services/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Services/ProviderSessionDirectory.ts @@ -45,6 +45,12 @@ export interface ProviderSessionDirectoryShape { binding: ProviderRuntimeBinding, ) => Effect.Effect; + /** + * Refresh only `lastSeenAt` for an existing binding, marking the session + * as recently active. No-op when no binding exists for the thread. + */ + readonly touch: (threadId: ThreadId) => Effect.Effect; + readonly getProvider: ( threadId: ThreadId, ) => Effect.Effect;