From 10a1d617d15ab146a951290de1f97f10578d547e Mon Sep 17 00:00:00 2001 From: Slate Rehm Date: Sat, 8 Aug 2026 16:56:33 -0500 Subject: [PATCH 1/2] fix: harden single-session lifecycle --- docs/configuration.md | 6 +- src/tools/registry.ts | 168 ++++++++++++---------- src/tools/session.ts | 51 +++++-- src/usage/activity-guard.ts | 14 +- tests/fixtures/settings-plugin/.gitignore | 1 + tests/unit/activity-guard.test.ts | 6 + tests/unit/server-binding.test.ts | 17 +++ 7 files changed, 172 insertions(+), 91 deletions(-) create mode 100644 tests/fixtures/settings-plugin/.gitignore diff --git a/docs/configuration.md b/docs/configuration.md index fecbe77..cda20c1 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -101,12 +101,12 @@ the package manager that supplied `installedPackage`. | `KNAP_RECONNECT_MS` | `2000` | Telemetry reconnect delay | Knapper writes default-profile telemetry to `KNAP_HOME/telemetry/events.jsonl`. -Each managed session uses the shared telemetry file in that directory. Knapper +The managed session uses `KNAP_HOME/telemetry/session.jsonl`. Knapper writes redacted tool audit events under `KNAP_HOME/audit`. Audit files use mode `0600` and have 14-day retention. -Session release and reset operations archive telemetry records. They store retained -records in the managed or quarantined root. +Session reset archives its telemetry in the quarantined root. Session release keeps +the telemetry file ready for the next agent that claims the same session. `LOG_LEVEL`, `RECONNECT_MS`, and `SCREENSHOT_DIR` are supported aliases. The `KNAP_` name takes precedence. diff --git a/src/tools/registry.ts b/src/tools/registry.ts index 358fca6..9b664be 100644 --- a/src/tools/registry.ts +++ b/src/tools/registry.ts @@ -371,86 +371,110 @@ export class ToolRegistry { const callRequestId = requestId(requestContext); const callArgs = args ?? {}; let queueMs = 0; - let admitted = false; let auditContext: AuditCallContext | undefined; let auditOutcome: "success" | "error" = "success"; let auditError: AuditErrorEnvelope | undefined; let completedOutcome: ToolOutcome | UobError | undefined; - return this.lock.run("exclusive", def.name, async () => { - try { - // Read the telemetry cursor after admission, not before: a call that - // waited in the queue would otherwise report every log line produced - // by the calls it was queued behind. - const sinceSeq = this.telemetry?.cursor ?? 0; - const ranAt = Date.now(); - admitted = true; - queueMs = ranAt - started; - await this.hooks.beforeInvoke?.(def, callArgs, requestContext); - auditContext = await this.hooks.contextProvider?.(callArgs, requestContext); - let outcome = await def.handler(callArgs); - completedOutcome = outcome; - if ( - this.telemetry && - def.annotations?.readOnlyHint !== true && - def.handlesOwnTelemetry !== true - ) { - outcome = withTelemetrySuffix(outcome, this.telemetry, sinceSeq); - } - this.logger.debug("tool ok", { - tool: def.name, - ms: Date.now() - ranAt, - queuedMs: ranAt - started, - }); - const result = normalize(outcome); - if (result.isError === true) { + try { + return await this.lock.run("exclusive", def.name, async () => { + try { + // Read the telemetry cursor after admission, not before: a call that + // waited in the queue would otherwise report every log line produced + // by the calls it was queued behind. + const sinceSeq = this.telemetry?.cursor ?? 0; + const ranAt = Date.now(); + queueMs = ranAt - started; + await this.hooks.beforeInvoke?.(def, callArgs, requestContext); + auditContext = await this.hooks.contextProvider?.(callArgs, requestContext); + let outcome = await def.handler(callArgs); + completedOutcome = outcome; + if ( + this.telemetry && + def.annotations?.readOnlyHint !== true && + def.handlesOwnTelemetry !== true + ) { + outcome = withTelemetrySuffix(outcome, this.telemetry, sinceSeq); + } + this.logger.debug("tool ok", { + tool: def.name, + ms: Date.now() - ranAt, + queuedMs: ranAt - started, + }); + const result = normalize(outcome); + if (result.isError === true) { + auditOutcome = "error"; + auditError = { + type: "ToolResultError", + code: "TOOL_ERROR", + message: "The tool call failed.", + retriable: false, + }; + } + return result; + } catch (e) { + const err = toUobError(e); + completedOutcome = err; auditOutcome = "error"; - auditError = { - type: "ToolResultError", - code: "TOOL_ERROR", - message: "The tool call failed.", - retriable: false, - }; - } - return result; - } catch (e) { - const err = toUobError(e); - completedOutcome = err; - auditOutcome = "error"; - auditError = errorEnvelope(err); - this.logger.warn("tool failed", { - tool: def.name, - code: err.code, - ms: Date.now() - started, - }); - return errorResult(err); - } finally { - if (completedOutcome !== undefined) { - try { - await this.hooks.afterInvoke?.(def, callArgs, requestContext, completedOutcome); - } catch (hookError) { - this.logger.warn("afterInvoke hook failed", { - tool: def.name, - error: hookError instanceof Error ? hookError.name : "UnknownHookError", - }); + auditError = errorEnvelope(err); + this.logger.warn("tool failed", { + tool: def.name, + code: err.code, + ms: Date.now() - started, + }); + return errorResult(err); + } finally { + if (completedOutcome !== undefined) { + try { + await this.hooks.afterInvoke?.(def, callArgs, requestContext, completedOutcome); + } catch (hookError) { + this.logger.warn("afterInvoke hook failed", { + tool: def.name, + error: hookError instanceof Error ? hookError.name : "UnknownHookError", + }); + } + } + if (this.audit !== false) { + this.queueAudit( + toolAuditEvent({ + timestamp, + requestId: callRequestId, + tool: def.name, + durationMs: Date.now() - started, + queueMs, + outcome: auditOutcome, + args: callArgs, + ...(auditContext ? { context: auditContext } : {}), + ...(auditError ? { error: auditError } : {}), + }), + ); } } - if (this.audit !== false) { - this.queueAudit( - toolAuditEvent({ - timestamp, - requestId: callRequestId, - tool: def.name, - durationMs: Date.now() - started, - queueMs: admitted ? queueMs : Date.now() - started, - outcome: auditOutcome, - args: callArgs, - ...(auditContext ? { context: auditContext } : {}), - ...(auditError ? { error: auditError } : {}), - }), - ); - } + }); + } catch (e) { + const err = toUobError(e); + auditOutcome = "error"; + auditError = errorEnvelope(err); + this.logger.warn("tool failed before admission", { + tool: def.name, + code: err.code, + ms: Date.now() - started, + }); + if (this.audit !== false) { + this.queueAudit( + toolAuditEvent({ + timestamp, + requestId: callRequestId, + tool: def.name, + durationMs: Date.now() - started, + queueMs: Date.now() - started, + outcome: auditOutcome, + args: callArgs, + error: auditError, + }), + ); } - }); + return errorResult(err); + } }) as never, ); } diff --git a/src/tools/session.ts b/src/tools/session.ts index c53cac8..db5f958 100644 --- a/src/tools/session.ts +++ b/src/tools/session.ts @@ -39,13 +39,11 @@ function compatible( return true; } -async function singletonDescriptor( +export function selectSingletonDescriptor( + descriptors: SessionDescriptor[], pluginSourceDir?: string, pluginId?: string, -): Promise { - const descriptors = (await listDescriptors()).filter( - (descriptor) => descriptor.readiness.phase !== "failed", - ); +): SessionDescriptor | undefined { if (descriptors.length === 0) return undefined; const descriptor = descriptors.at(-1) as SessionDescriptor; if (!compatible(descriptor, pluginSourceDir, pluginId)) { @@ -61,22 +59,36 @@ async function singletonDescriptor( return descriptor; } +async function singletonDescriptor( + pluginSourceDir?: string, + pluginId?: string, +): Promise { + return selectSingletonDescriptor(await listDescriptors(), pluginSourceDir, pluginId); +} + async function makeReady( ctx: ServerContext, descriptor: SessionDescriptor, timeoutMs?: number, ): Promise { + const startedAt = Date.now(); + const remainingOptions = (): { timeoutMs?: number } => + timeoutMs === undefined ? {} : { timeoutMs: Math.max(1, timeoutMs - (Date.now() - startedAt)) }; let next = descriptor; if (next.readiness.phase === "starting") { - next = await waitSession(next.key, timeoutMs !== undefined ? { timeoutMs } : {}); - } else if (next.readiness.phase === "stopped" || (await sessionState(next)) !== "live") { + next = await waitSession(next.key, remainingOptions()); + } else if ( + next.readiness.phase === "failed" || + next.readiness.phase === "stopped" || + (await sessionState(next)) !== "live" + ) { const restarted = await restartSession(next.key, { logger: ctx.logger.child("session"), - ...(timeoutMs !== undefined ? { timeoutMs } : {}), + ...remainingOptions(), }); next = restarted.descriptor; if (next.readiness.phase === "starting") { - next = await waitSession(next.key, timeoutMs !== undefined ? { timeoutMs } : {}); + next = await waitSession(next.key, remainingOptions()); } } await ctx.bindSession(next); @@ -240,11 +252,19 @@ export function registerSessionTools(ctx: ServerContext): void { }, handler: async (args) => { const timeoutMs = typeof args.timeoutMs === "number" ? args.timeoutMs : undefined; + const startedAt = Date.now(); + const remainingTimeout = (): number | undefined => + timeoutMs === undefined ? undefined : Math.max(1, timeoutMs - (Date.now() - startedAt)); const descriptors = await listDescriptors(); const previous = ctx.currentSessionKey ?? descriptors.at(-1)?.key; let quarantinedPath: string | undefined; + let archivedTelemetry: string | undefined; if (previous !== undefined) { - const stopped = await stopSession(previous, timeoutMs !== undefined ? { timeoutMs } : {}); + const stopTimeout = remainingTimeout(); + const stopped = await stopSession( + previous, + stopTimeout !== undefined ? { timeoutMs: stopTimeout } : {}, + ); if (stopped.state === "quitFailed") { throw new UobError("TIMEOUT", `Session ${previous} did not stop.`, { remediation: "Retry after the managed Obsidian process stops.", @@ -255,12 +275,21 @@ export function registerSessionTools(ctx: ServerContext): void { await ctx.bindDefault(); ctx.currentSessionKey = undefined; ctx.targetKind = undefined; - const descriptor = await openIsolated(ctx, args); + ctx.selectTelemetry("default"); + if (quarantinedPath !== undefined) { + archivedTelemetry = await ctx.archiveTelemetry("session", quarantinedPath); + } + const nextTimeout = remainingTimeout(); + const descriptor = await openIsolated(ctx, { + ...args, + ...(nextTimeout !== undefined ? { timeoutMs: nextTimeout } : {}), + }); return { text: `Fresh isolated session ${descriptor.key} is ready.`, json: { reset: previous ?? null, quarantinedPath: quarantinedPath ?? null, + archivedTelemetry: archivedTelemetry ?? null, ...publicSummary(descriptor), }, }; diff --git a/src/usage/activity-guard.ts b/src/usage/activity-guard.ts index 636d58d..dbbe43d 100644 --- a/src/usage/activity-guard.ts +++ b/src/usage/activity-guard.ts @@ -64,10 +64,14 @@ export class ActivityGuard { this.now = opts.now ?? (() => new Date()); this.pid = opts.pid ?? process.pid; this.host = opts.hostname ?? hostname(); - if (!Number.isFinite(opts.idleTimeoutMs) || opts.idleTimeoutMs <= 0) { - throw new UobError("INVALID_ARGUMENT", "The activity idle timeout must be positive.", { - remediation: "Set KNAP_ACTIVITY_IDLE_MS to a positive number of milliseconds.", - }); + if (!Number.isFinite(opts.idleTimeoutMs) || opts.idleTimeoutMs < 1_000) { + throw new UobError( + "INVALID_ARGUMENT", + "The activity idle timeout must be at least 1000 ms.", + { + remediation: "Set KNAP_ACTIVITY_IDLE_MS to at least 1000 milliseconds.", + }, + ); } } @@ -118,7 +122,7 @@ export class ActivityGuard { private startHeartbeat(): void { if (this.heartbeatTimer !== undefined) clearInterval(this.heartbeatTimer); - const intervalMs = Math.max(1_000, Math.floor(this.opts.idleTimeoutMs / 3)); + const intervalMs = Math.max(250, Math.floor(this.opts.idleTimeoutMs / 3)); this.heartbeatTimer = setInterval( () => void this.heartbeat().catch(() => undefined), intervalMs, diff --git a/tests/fixtures/settings-plugin/.gitignore b/tests/fixtures/settings-plugin/.gitignore new file mode 100644 index 0000000..2d46485 --- /dev/null +++ b/tests/fixtures/settings-plugin/.gitignore @@ -0,0 +1 @@ +data.json diff --git a/tests/unit/activity-guard.test.ts b/tests/unit/activity-guard.test.ts index 2cfe49f..038eb20 100644 --- a/tests/unit/activity-guard.test.ts +++ b/tests/unit/activity-guard.test.ts @@ -17,6 +17,12 @@ afterEach(async () => { }); describe("ActivityGuard", () => { + it("rejects an idle window shorter than the heartbeat floor", () => { + expect(() => new ActivityGuard({ idleTimeoutMs: 999, env })).toThrow( + "activity idle timeout must be at least 1000 ms", + ); + }); + it("starts free, tracks an operation, and does not renew from status", async () => { let now = new Date("2026-08-08T12:00:00Z"); const first = new ActivityGuard({ diff --git a/tests/unit/server-binding.test.ts b/tests/unit/server-binding.test.ts index 1874aa4..9d5967d 100644 --- a/tests/unit/server-binding.test.ts +++ b/tests/unit/server-binding.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import { loadConfig } from "../../src/config.js"; import { applySessionConfig } from "../../src/server.js"; import type { SessionDescriptor } from "../../src/session/descriptor.js"; +import { selectSingletonDescriptor } from "../../src/tools/session.js"; function descriptor(): SessionDescriptor { return { @@ -58,3 +59,19 @@ describe("applySessionConfig", () => { expect(() => applySessionConfig(loadConfig({}, {}), pending)).toThrow(/not ready/); }); }); + +describe("selectSingletonDescriptor", () => { + it("retains a failed descriptor so recovery cannot create a second target", () => { + const failed = descriptor(); + failed.readiness = { + phase: "failed", + failedAt: "2026-08-04T12:00:02.000Z", + code: "TIMEOUT", + message: "startup verification timed out", + remediation: "Retry the managed session.", + fixedBy: "obsidian_session_reset", + }; + + expect(selectSingletonDescriptor([failed])).toBe(failed); + }); +}); From 4b547f2a6b12790b49cd8354088ab48840f294a4 Mon Sep 17 00:00:00 2001 From: Slate Rehm Date: Sat, 8 Aug 2026 16:58:57 -0500 Subject: [PATCH 2/2] fix: preserve session timeout deadlines --- src/tools/session.ts | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/src/tools/session.ts b/src/tools/session.ts index db5f958..0cb78e4 100644 --- a/src/tools/session.ts +++ b/src/tools/session.ts @@ -69,11 +69,10 @@ async function singletonDescriptor( async function makeReady( ctx: ServerContext, descriptor: SessionDescriptor, - timeoutMs?: number, + deadline?: number, ): Promise { - const startedAt = Date.now(); const remainingOptions = (): { timeoutMs?: number } => - timeoutMs === undefined ? {} : { timeoutMs: Math.max(1, timeoutMs - (Date.now() - startedAt)) }; + deadline === undefined ? {} : { timeoutMs: Math.max(1, deadline - Date.now()) }; let next = descriptor; if (next.readiness.phase === "starting") { next = await waitSession(next.key, remainingOptions()); @@ -99,11 +98,16 @@ async function makeReady( async function openIsolated( ctx: ServerContext, args: Record, + inheritedDeadline?: number, ): Promise { const pluginSourceDir = typeof args.pluginSourceDir === "string" ? args.pluginSourceDir : undefined; const pluginId = typeof args.pluginId === "string" ? args.pluginId : undefined; const timeoutMs = typeof args.timeoutMs === "number" ? args.timeoutMs : undefined; + const deadline = + inheritedDeadline ?? (timeoutMs === undefined ? undefined : Date.now() + timeoutMs); + const timeoutOptions = (): { timeoutMs?: number } => + deadline === undefined ? {} : { timeoutMs: Math.max(1, deadline - Date.now()) }; let descriptor = await singletonDescriptor(pluginSourceDir, pluginId); if (descriptor === undefined) { descriptor = await createSession({ @@ -112,10 +116,10 @@ async function openIsolated( ...(typeof args.label === "string" ? { label: args.label } : {}), ...(pluginSourceDir !== undefined ? { pluginSourceDir } : {}), ...(pluginId !== undefined ? { pluginId } : {}), - ...(timeoutMs !== undefined ? { timeoutMs } : {}), + ...timeoutOptions(), }); } - return makeReady(ctx, descriptor, timeoutMs); + return makeReady(ctx, descriptor, deadline); } export function registerSessionTools(ctx: ServerContext): void { @@ -252,9 +256,9 @@ export function registerSessionTools(ctx: ServerContext): void { }, handler: async (args) => { const timeoutMs = typeof args.timeoutMs === "number" ? args.timeoutMs : undefined; - const startedAt = Date.now(); + const deadline = timeoutMs === undefined ? undefined : Date.now() + timeoutMs; const remainingTimeout = (): number | undefined => - timeoutMs === undefined ? undefined : Math.max(1, timeoutMs - (Date.now() - startedAt)); + deadline === undefined ? undefined : Math.max(1, deadline - Date.now()); const descriptors = await listDescriptors(); const previous = ctx.currentSessionKey ?? descriptors.at(-1)?.key; let quarantinedPath: string | undefined; @@ -279,11 +283,7 @@ export function registerSessionTools(ctx: ServerContext): void { if (quarantinedPath !== undefined) { archivedTelemetry = await ctx.archiveTelemetry("session", quarantinedPath); } - const nextTimeout = remainingTimeout(); - const descriptor = await openIsolated(ctx, { - ...args, - ...(nextTimeout !== undefined ? { timeoutMs: nextTimeout } : {}), - }); + const descriptor = await openIsolated(ctx, args, deadline); return { text: `Fresh isolated session ${descriptor.key} is ready.`, json: {