diff --git a/.changeset/assistant-visibility-reconciliation.md b/.changeset/assistant-visibility-reconciliation.md new file mode 100644 index 000000000..5b5f388e3 --- /dev/null +++ b/.changeset/assistant-visibility-reconciliation.md @@ -0,0 +1,5 @@ +--- +"@sapiom/harness": patch +--- + +Keep continuation visibility consistent across actions and events while preserving End during storage failures. diff --git a/packages/harness/docs/assistant-history.md b/packages/harness/docs/assistant-history.md index 1941395b6..caa5790f1 100644 --- a/packages/harness/docs/assistant-history.md +++ b/packages/harness/docs/assistant-history.md @@ -46,6 +46,7 @@ lifecycle and idle checkpoints can flush immediately. Native reads are capped at 16 MiB and have a three-second timeout. Failure preserves the previous record, so a successful GET can return an older `capturedAt`. Disposal and authority loss cancel capture; opening a retained record never resumes execution. + # Selected native availability The Resume coordinator claims the same operation slot as Attach and inspection. @@ -200,3 +201,9 @@ After a verified Continue lifecycle conflict, Review latest record fetches and displays saved history without creating work. A separate Start a new continuation action replaces the rejected request only if its saved tuple is still unchanged. Uncertain failures and reloads continue to reconcile the original operation. + +Pending-child admission also covers macros, Canvas and Terminal connections. +Visibility changes publish on the existing Assistant state revision clock, and +an older read cannot overwrite newer completed evidence. An unreadable index +blocks discovery and execution while exact authorized End remains available; +previously verified pending children remain private during that failure. diff --git a/packages/harness/src/core/assistant-state-projection.test.ts b/packages/harness/src/core/assistant-state-projection.test.ts index 93053a03f..754d1c3cf 100644 --- a/packages/harness/src/core/assistant-state-projection.test.ts +++ b/packages/harness/src/core/assistant-state-projection.test.ts @@ -117,6 +117,33 @@ it("keeps private continuation runtimes and lifecycle headers out of every snaps projection.dispose(); }); +it("publishes visibility changes on the same clock without a runtime or lifecycle event", () => { + let visible = false; + const publish = vi.fn(); + const projection = createAssistantStateProjection( + { + getAssistantState: () => runtime, + subscribeAssistantState: () => () => {}, + }, + { snapshot: () => [state], subscribe: () => () => {} }, + publish, + () => visible, + ); + const before = projection.get(); + visible = true; + projection.invalidate(); + const shown = publish.mock.lastCall![0] as AssistantStateSnapshot; + expect(shown.revision).toBeGreaterThan(before.revision); + expect(shown.lifecycles).toEqual([state]); + visible = false; + projection.invalidate(); + expect(publish.mock.lastCall![0]).toMatchObject({ + revision: shown.revision + 1, + lifecycles: [], + }); + projection.dispose(); +}); + it.each([ [{ ...state, lease: "private" }], [{ ...state, execution: "enabled" }], diff --git a/packages/harness/src/core/assistant-state-projection.ts b/packages/harness/src/core/assistant-state-projection.ts index 66f7ce6b1..f8a82f15c 100644 --- a/packages/harness/src/core/assistant-state-projection.ts +++ b/packages/harness/src/core/assistant-state-projection.ts @@ -38,6 +38,7 @@ export function createAssistantStateProjection( const stopLifecycle = lifecycle.subscribe(changed); return { get, + invalidate: changed, dispose: () => { stopHost(); stopLifecycle(); diff --git a/packages/harness/src/server/assistant-lifecycle-wiring.test.ts b/packages/harness/src/server/assistant-lifecycle-wiring.test.ts index e48cd6780..719e660f4 100644 --- a/packages/harness/src/server/assistant-lifecycle-wiring.test.ts +++ b/packages/harness/src/server/assistant-lifecycle-wiring.test.ts @@ -49,6 +49,7 @@ import { import type { AssistantGrant } from "../core/assistant-access.js"; import type { HostedOpenCode } from "../core/opencode-host.js"; import type { AssistantHistoryEntry } from "../shared/assistant-history.js"; +import type { AssistantStateSnapshot } from "../shared/assistant-state.js"; import type { HarnessAdapter, HarnessSession, @@ -252,6 +253,29 @@ async function entries(): Promise { async function header(id = parentId) { return json(`/opencode/${id}/lifecycle`); } +async function socketSnapshot(): Promise { + const observer = new WebSocket( + `ws://127.0.0.1:${studio!.port}/ws/events?token=${token}`, + ); + return new Promise((resolve, reject) => { + observer.once("message", (raw) => { + observer.terminate(); + resolve(JSON.parse(raw.toString()).snapshot as AssistantStateSnapshot); + }); + observer.once("error", reject); + }); +} +async function terminalRejection(id: string, credential = token) { + const observer = new WebSocket( + `ws://127.0.0.1:${studio!.port}/ws/terminal?session=${id}&token=${credential}`, + ); + return new Promise<{ code: number; reason: string }>((resolve, reject) => { + observer.once("close", (code, reason) => + resolve({ code, reason: reason.toString() }), + ); + observer.once("error", reject); + }); +} function owner(): ActivatedAssistantContextRuntime { return { start: vi.fn(nativeStart), assertAvailable: vi.fn(async () => {}) }; } @@ -805,6 +829,20 @@ it("shares retained context across Continue, later Send and Resume while Termina }, 20_000); it("hides an allocated but unprepared child across reboot and blocks attach before another native start", async () => { + const workflow = join(cwd, "example-agent"); + await mkdir(workflow); + await writeFile( + join(workflow, "sapiom.json"), + JSON.stringify({ definitionId: null }), + ); + await writeFile( + join(workflow, "package.json"), + JSON.stringify({ name: "example-agent" }), + ); + const canvas = join(cwd, ".sapiom", "canvas"); + await mkdir(canvas, { recursive: true }); + await writeFile(join(canvas, "index.html"), "Authored canvas"); + await writeFile(join(canvas, "asset.txt"), "Authored asset"); const activated = { ...owner(), loadGuidance: async () => { @@ -822,6 +860,9 @@ it("hides an allocated but unprepared child across reboot and blocks attach befo const receipt = (await stores.continuations.read(binding, operationId))!; expect(receipt.phase).toBe("associated"); const assertPrivate = async () => { + expect(await json("/api/workflows")).toContainEqual( + expect.objectContaining({ path: workflow }), + ); const state = await json("/api/state"); expect(JSON.stringify(state)).not.toContain(receipt.childStudioId); expect(await json("/api/sessions")).not.toContainEqual( @@ -847,6 +888,57 @@ it("hides an allocated but unprepared child across reboot and blocks attach befo ).status, `${method} ${suffix}`, ).toBe(404); + for (const [path, method, body] of [ + [ + "/api/macros/describe/run", + "POST", + { + harnessSessionId: receipt.childStudioId, + workflowPath: workflow, + subject: "Must not run", + }, + ], + [ + "/api/macros/run_local/run", + "POST", + { harnessSessionId: receipt.childStudioId, workflowPath: workflow }, + ], + [ + "/api/macros/visualize/run", + "POST", + { harnessSessionId: receipt.childStudioId }, + ], + [`/api/canvas/${receipt.childStudioId}/render`, "POST", {}], + [`/canvas/${receipt.childStudioId}`, "GET", undefined], + [`/canvas/${receipt.childStudioId}/asset.txt`, "GET", undefined], + ] as const) + expect( + (await request(path, method, body)).status, + `${method} ${path}`, + ).toBe(404); + expect((await request(`/canvas/${parentId}`)).status).toBe(200); + expect((await request(`/canvas/${parentId}/asset.txt`)).status).toBe(200); + expect(await terminalRejection(receipt.childStudioId)).toEqual({ + code: 4004, + reason: "session not found", + }); + expect( + await terminalRejection(receipt.childStudioId, "wrong-token"), + ).toEqual({ code: 4001, reason: "unauthorized" }); + const unavailable = vi + .spyOn(AssistantContinuationStore.prototype, "readChild") + .mockRejectedValue(new Error("Index unavailable")); + try { + expect( + (await request(`/api/sessions/${receipt.childStudioId}/record`)).status, + ).toBe(404); + expect( + (await request(`/api/sessions/${receipt.childStudioId}`, "DELETE")) + .status, + ).toBe(404); + } finally { + unavailable.mockRestore(); + } expect(JSON.stringify(events)).not.toContain(receipt.childStudioId); }; await assertPrivate(); @@ -888,8 +980,201 @@ it("hides an allocated but unprepared child across reboot and blocks attach befo ).toBe(404); expect(launches).toHaveLength(before); expect(terminal).not.toHaveBeenCalled(); + expect(background).not.toHaveBeenCalled(); expect( calls.filter(({ path }) => path.endsWith("/prompt_async")), ).toHaveLength(0); expect(providerRequests).toEqual([]); }, 20_000); + +it("ends only the selected live Terminal despite an unreadable continuation index", async () => { + await boot(owner()); + const selected = await studio!.sessionManager.create({ + cwd, + harness: "claude-code", + }); + const neighbor = await studio!.sessionManager.create({ + cwd, + harness: "claude-code", + }); + const directory = join(root, "assistant-sessions", selected.id); + await mkdir(directory, { recursive: true }); + await writeFile(join(directory, "continuation-preparation.json"), "not-json"); + expect(await json("/api/sessions")).not.toContainEqual( + expect.objectContaining({ id: selected.id }), + ); + expect( + ( + await request(`/api/sessions/${selected.id}/input`, "POST", { + text: "must not run", + }) + ).status, + ).toBe(404); + const response = await request(`/api/sessions/${selected.id}`, "DELETE"); + expect(response.status).toBe(200); + expect(studio!.sessionManager.isLive(selected.id)).toBe(false); + expect(studio!.sessionManager.isLive(neighbor.id)).toBe(true); + expect(studio!.sessionManager.get(selected.id)).toBeDefined(); +}, 20_000); + +it("keeps a prepared continuation visible after an older pending read finishes", async () => { + const activated = { + ...owner(), + loadGuidance: async () => { + throw new Error("Unavailable"); + }, + }; + await boot(activated); + const operationId = randomUUID(); + const input = { expectedRevision: 1, expectedRecordRevision: 1, operationId }; + expect( + ( + await request( + `/api/sessions/${parentId}/assistant/continue`, + "POST", + input, + ) + ).status, + ).toBe(503); + const receipt = (await stores.continuations.read(binding, operationId))!; + Object.assign(activated, { loadGuidance: undefined }); + const read = AssistantContinuationStore.prototype.readChild; + let enter!: () => void, + release!: () => void, + held = false; + const entered = new Promise((resolve) => { + enter = resolve; + }); + const wait = new Promise((resolve) => { + release = resolve; + }); + const spy = vi + .spyOn(AssistantContinuationStore.prototype, "readChild") + .mockImplementation(async function (this: AssistantContinuationStore, id) { + const saved = await read.call(this, id); + if ( + !held && + id === receipt.childStudioId && + saved?.phase !== "prepared" + ) { + held = true; + enter(); + await wait; + } + return saved; + }); + let pending: Promise | undefined; + try { + pending = request(`/api/sessions/${receipt.childStudioId}/record`); + await entered; + const completed = await json( + `/api/sessions/${parentId}/assistant/continue`, + "POST", + input, + ); + expect(completed.session.id).toBe(receipt.childStudioId); + expect((await stores.continuations.read(binding, operationId))!.phase).toBe( + "prepared", + ); + release(); + expect((await pending).status).toBe(404); + expect((await socketSnapshot()).lifecycles).toContainEqual( + expect.objectContaining({ harnessSessionId: receipt.childStudioId }), + ); + } finally { + release(); + await pending; + spy.mockRestore(); + } +}, 20_000); + +it("publishes a new Assistant revision when visibility alone changes", async () => { + await boot(owner()); + const before = await socketSnapshot(); + expect(before.lifecycles).toContainEqual( + expect.objectContaining({ harnessSessionId: parentId }), + ); + const read = AssistantContinuationStore.prototype.readChild; + let unreadable = true; + const spy = vi + .spyOn(AssistantContinuationStore.prototype, "readChild") + .mockImplementation(async function (this: AssistantContinuationStore, id) { + if (id === parentId && unreadable) throw new Error("Read unavailable"); + return read.call(this, id); + }); + try { + await json("/api/sessions"); + await vi.waitFor(() => + expect(events).toContainEqual( + expect.objectContaining({ + type: "assistant.state", + snapshot: expect.objectContaining({ + revision: before.revision + 1, + lifecycles: [], + }), + }), + ), + ); + const hidden = await socketSnapshot(); + unreadable = false; + await json("/api/sessions"); + await vi.waitFor(() => + expect(events).toContainEqual( + expect.objectContaining({ + type: "assistant.state", + snapshot: expect.objectContaining({ + revision: hidden.revision + 1, + lifecycles: before.lifecycles, + }), + }), + ), + ); + } finally { + spy.mockRestore(); + } +}, 20_000); + +it("publishes completed visibility reads while a newer refresh is still pending", async () => { + await boot(owner()); + const read = AssistantContinuationStore.prototype.readChild; + const entered: (() => void)[] = [], + release: (() => void)[] = []; + const starts = [0, 1].map( + () => new Promise((resolve) => entered.push(resolve)), + ); + const gates = [0, 1].map( + () => new Promise((resolve) => release.push(resolve)), + ); + let calls = 0; + const spy = vi + .spyOn(AssistantContinuationStore.prototype, "readChild") + .mockImplementation(async function (this: AssistantContinuationStore, id) { + if (id !== parentId || calls > 1) return read.call(this, id); + const index = calls++; + entered[index]!(); + await gates[index]; + if (index === 0) throw new Error("Index unavailable"); + return read.call(this, id); + }); + const pending: Promise[] = []; + try { + pending.push(request("/api/sessions")); + await starts[0]; + pending.push(request("/api/sessions")); + await starts[1]; + release[0]!(); + expect(await (await pending[0]!).json()).toEqual([]); + expect((await socketSnapshot()).lifecycles).toEqual([]); + release[1]!(); + expect(await (await pending[1]!).json()).toContainEqual( + expect.objectContaining({ id: parentId }), + ); + expect((await socketSnapshot()).lifecycles).toContainEqual( + expect.objectContaining({ harnessSessionId: parentId }), + ); + } finally { + release.forEach((resolve) => resolve()); + await Promise.allSettled(pending); + spy.mockRestore(); + } +}, 20_000); diff --git a/packages/harness/src/server/index.ts b/packages/harness/src/server/index.ts index 8a13846b2..f818aa47d 100644 --- a/packages/harness/src/server/index.ts +++ b/packages/harness/src/server/index.ts @@ -158,7 +158,10 @@ import { import { resolveManifestName, } from "../core/definition-name.js"; -import { createBootTokenMiddleware } from "./auth.js"; +import { + createBootTokenMiddleware, + timingSafeEqualString, +} from "./auth.js"; import { createOpenCodeRouter } from "./opencode.js"; import { createApiKeyProvider, @@ -3558,11 +3561,38 @@ export const startServer = async ( const assistantSessions = new AssistantSessionStore(statePaths.root); const assistantRecords = new AssistantRecordStore(statePaths.root); const assistantContinuations = new AssistantContinuationStore(statePaths.root); + // Keep the last positive pending proof through read failures. Unknown/corrupt + // indexes still deny discovery and execution, but cannot block ordinary End. + const pendingAssistantSessions = new Set(); + const publishedVisibilityReads = new Map(); + let visibilityReadRevision = 0; + const visibilityProjection: { invalidate?: () => void } = {}; const isAssistantSessionVisible = async (id: string): Promise => { - const visible = await assistantContinuations.canAttach(id); - if (visible) hiddenAssistantSessions.delete(id); - else hiddenAssistantSessions.add(id); - return visible; + const read = ++visibilityReadRevision; + let status: "visible" | "pending" | "unavailable"; + try { + const receipt = await assistantContinuations.readChild(id); + status = + receipt === null || receipt.phase === "prepared" ? "visible" : "pending"; + } catch { + status = "unavailable"; + } + // Publish completed evidence immediately; a later in-flight read is not + // evidence yet, and an older completion cannot replace a newer publication. + if (read > (publishedVisibilityReads.get(id) ?? 0)) { + publishedVisibilityReads.set(id, read); + const wasHidden = hiddenAssistantSessions.has(id); + if (status === "visible") { + hiddenAssistantSessions.delete(id); + pendingAssistantSessions.delete(id); + } else { + hiddenAssistantSessions.add(id); + if (status === "pending") pendingAssistantSessions.add(id); + } + if (wasHidden !== hiddenAssistantSessions.has(id)) + visibilityProjection.invalidate?.(); + } + return status === "visible"; }; const restoredSessionIds = sessionManager.list().map((session) => session.id); for (let offset = 0; offset < restoredSessionIds.length; offset += 8) @@ -3615,6 +3645,7 @@ export const startServer = async ( bus.publish({ type: "assistant.state", snapshot }), (id) => !hiddenAssistantSessions.has(id), ); + visibilityProjection.invalidate = assistantProjection.invalidate; const getAssistantState = assistantProjection.get; const assistantContext = createAssistantContextRuntime({ activated: options.assistantContextRuntime, @@ -3688,6 +3719,35 @@ export const startServer = async ( createBootTokenMiddleware(options.bootToken), express.json({ limit: JSON_BODY_LIMIT_BYTES }), ); + // These session actions live outside /sessions/:id. Resolve their body/path + // target through the same admission check before any task or file operation. + const requireVisibleSession = ( + select: (req: express.Request) => unknown, + ): express.RequestHandler => + async (req, res, next) => { + try { + const id = select(req); + if ( + typeof id === "string" && + sessionManager.get(id) && + !(await isAssistantSessionVisible(id)) + ) { + res.status(404).json({ error: "session not found" }); + return; + } + next(); + } catch (error) { + next(error); + } + }; + app.post( + "/api/macros/:id/run", + requireVisibleSession((req) => req.body?.harnessSessionId), + ); + app.post( + "/api/canvas/:sessionId/render", + requireVisibleSession((req) => req.params.sessionId), + ); app.get("/api/assistant/access", (_req, res) => { res.setHeader("Cache-Control", "no-store"); res.json(assistantAccess.getBrowserState()); @@ -3716,6 +3776,7 @@ export const startServer = async ( "/api", createRestRouter({ isSessionVisible: isAssistantSessionVisible, + isSessionEndAllowed: (id) => !pendingAssistantSessions.has(id), getAssistantState, endSession: (id) => assistantEnd.end(id), sessionManager, @@ -4351,6 +4412,10 @@ export const startServer = async ( // Canvas is intentionally unauthenticated (see canvas.ts) — served straight // off the session's cwd, no boot token required. + app.use( + "/canvas/:harnessSessionId", + requireVisibleSession((req) => req.params.harnessSessionId), + ); app.use( createCanvasRouter((harnessSessionId) => { const session = sessionManager.get(harnessSessionId); @@ -4381,14 +4446,26 @@ export const startServer = async ( const terminalWss = new WebSocketServer({ noServer: true }); const eventsWss = new WebSocketServer({ noServer: true }); + const terminalConnection = createTerminalWebSocketHandler( + sessionManager, + options.bootToken, + ); attachWebSocketRouters(httpServer, [ { path: "/ws/terminal", wss: terminalWss, - onConnection: createTerminalWebSocketHandler( - sessionManager, - options.bootToken, - ), + onConnection: (ws, req, params) => { + // The cache is hydrated before listening and before child allocation. + // Preserve auth precedence without disk I/O ahead of the byte handler. + if ( + timingSafeEqualString(params.get("token") ?? "", options.bootToken) && + hiddenAssistantSessions.has(params.get("session") ?? "") + ) { + ws.close(4004, "session not found"); + return; + } + terminalConnection(ws, req, params); + }, }, { path: "/ws/events", diff --git a/packages/harness/src/server/rest.test.ts b/packages/harness/src/server/rest.test.ts index 5b22d9a0b..fe76c0c8b 100644 --- a/packages/harness/src/server/rest.test.ts +++ b/packages/harness/src/server/rest.test.ts @@ -404,6 +404,47 @@ describe("createRestRouter", () => { }); describe("DELETE /sessions/:id", () => { + it("requests End synchronously without waiting on an unreadable visibility index", async () => { + const sessionManager = fakeSessionManager([exitedSession()]); + const endSession = vi.fn().mockResolvedValue({ ok: true }); + const isSessionVisible = vi.fn(() => new Promise(() => {})); + start({ + sessionManager, + endSession, + isSessionVisible, + isSessionEndAllowed: () => true, + }); + const res = await fetch(`${baseUrl}/sessions/sess-1`, { + method: "DELETE", + headers: TOKEN_HEADER, + signal: AbortSignal.timeout(1_000), + }); + expect(res.status).toBe(200); + expect(endSession).toHaveBeenCalledExactlyOnceWith("sess-1"); + expect(isSessionVisible).not.toHaveBeenCalled(); + }); + + it("keeps positively pending children private while gating other methods and descendants", async () => { + const sessionManager = fakeSessionManager([exitedSession()]); + const endSession = vi.fn(); + const isSessionVisible = vi.fn().mockResolvedValue(false); + const isSessionEndAllowed = vi.fn().mockReturnValue(false); + start({ sessionManager, endSession, isSessionVisible, isSessionEndAllowed }); + const call = (suffix: string, method: string) => + fetch(`${baseUrl}/sessions/sess-1${suffix}`, { + method, + headers: TOKEN_HEADER, + }); + expect((await call("", "DELETE")).status).toBe(404); + expect(isSessionVisible).not.toHaveBeenCalled(); + expect(endSession).not.toHaveBeenCalled(); + isSessionEndAllowed.mockReturnValue(true); + expect((await call("/input", "POST")).status).toBe(404); + expect((await call("/record", "DELETE")).status).toBe(404); + expect(isSessionVisible).toHaveBeenCalledTimes(2); + expect(sessionManager.submitInput).not.toHaveBeenCalled(); + }); + it("keeps the legacy close path when no End coordinator is installed", async () => { const sessionManager = fakeSessionManager([exitedSession()]); start({ sessionManager }); diff --git a/packages/harness/src/server/rest.ts b/packages/harness/src/server/rest.ts index 4d931b075..0bd8ec1e0 100644 --- a/packages/harness/src/server/rest.ts +++ b/packages/harness/src/server/rest.ts @@ -200,6 +200,8 @@ async function agentHoldsConversation( export interface RestRouterOptions { /** Keep unfinished continuation allocations private without removing internal registry rows. */ isSessionVisible?: (id: string) => Promise; + /** Cached pending proof only: exact boot-authorized End must not wait on index I/O. */ + isSessionEndAllowed?: (id: string) => boolean; getAssistantState?: () => AssistantStateSnapshot; /** Complete selected-session End, including native and Terminal cleanup. * Uses the route's boot authorization independently of Assistant grant expiry. */ @@ -370,7 +372,14 @@ export function createRestRouter(options: RestRouterOptions): Router { }; router.use("/sessions/:id", async (req, res, next) => { try { - if (options.isSessionVisible && sessionManager.get(req.params.id) && !(await options.isSessionVisible(req.params.id))) { + const exactEnd = req.method === "DELETE" && req.path === "/"; + const allowed = + exactEnd && options.isSessionEndAllowed + ? options.isSessionEndAllowed(req.params.id) + : !options.isSessionVisible || + !sessionManager.get(req.params.id) || + (await options.isSessionVisible(req.params.id)); + if (!allowed) { res.status(404).json({ error: "session not found" }); return; }