From dd36b3fab626dfa4cf0079fbf8cc9a8bf087776a Mon Sep 17 00:00:00 2001 From: SummerC0zyR0ck Date: Wed, 16 Sep 2026 09:13:13 +0000 Subject: [PATCH 1/5] fix(desktop): invalidate artifact previews after deletion --- .../managed-artifact-preview.test.ts | 71 +++++++++++++++++- .../src/main/managed-artifact-preview.ts | 23 +++++- .../main/runtime-host-artifacts-ipc-main.ts | 2 + apps/desktop/src/main/runtime-host-boot.ts | 8 ++ apps/desktop/src/main/runtime-host-client.ts | 6 ++ .../src/__tests__/acp-stdio-server.test.ts | 1 + .../runtime-host-cli-context.test.ts | 9 +++ .../__tests__/runtime-host-onboarding.test.ts | 1 + .../tui-mcp-remote-publication.test.ts | 1 + .../__tests__/artifact-coordinator.test.ts | 69 +++++++++++++++++ .../src/__tests__/artifact-protocol.test.ts | 59 +++++++++++++++ .../__tests__/artifact-two-client-uds.test.ts | 37 +++++++++- .../src/__tests__/connection-session.test.ts | 60 +++++++++++++++ .../src/__tests__/host-change-feed.test.ts | 38 +++++++++- .../src/__tests__/protocol.test.ts | 4 + .../__tests__/reconnecting-connection.test.ts | 74 +++++++++++++++++++ .../session-retirement-coordinator.test.ts | 50 +++++++++++++ .../runtime-host/src/client/connection.ts | 22 ++++++ .../src/client/reconnecting-connection.ts | 10 +++ .../src/protocol/artifact-change.ts | 62 ++++++++++++++++ packages/runtime-host/src/protocol/index.ts | 7 +- .../src/server/artifact-coordinator.ts | 4 + .../src/server/connection-session.ts | 9 +++ .../src/server/execution-composition.ts | 3 + .../src/server/host-change-feed.ts | 29 ++++++-- .../server/session-retirement-coordinator.ts | 4 + .../server/session-revision-coordinator.ts | 4 + .../src/server/session-sidecar-purge.ts | 8 +- 28 files changed, 659 insertions(+), 16 deletions(-) create mode 100644 packages/runtime-host/src/protocol/artifact-change.ts diff --git a/apps/desktop/src/main/__tests__/managed-artifact-preview.test.ts b/apps/desktop/src/main/__tests__/managed-artifact-preview.test.ts index 61c99344a7..c0481c51bc 100644 --- a/apps/desktop/src/main/__tests__/managed-artifact-preview.test.ts +++ b/apps/desktop/src/main/__tests__/managed-artifact-preview.test.ts @@ -136,18 +136,81 @@ test('close and cancellation during a stream cannot publish a live endpoint', as } }); -test('bounds concurrent preparations before allocating buffers or ports', async () => { +test('bounds previews per session instead of starving another session', async () => { const service = new ManagedArtifactPreview(); let resume!: () => void; const gate = new Promise((resolve) => { resume = resolve; }); const source = client(); - const slow = { ...source, getArtifact: async (s: string, a: string) => { await gate; return source.getArtifact(s, a); } }; + const slow = { ...source, getArtifact: async () => { await gate; return source.getArtifact('s1', 'a1'); } }; const pending = Array.from({ length: 16 }, () => service.prepare('h', slow, 's1', 'a1')); try { await assert.rejects(service.prepare('h', slow, 's1', 'a1'), /Too many/); + const other = service.prepare('h', client('other'), 's2', 'a1'); resume(); - assert.equal((await Promise.all(pending)).length, 16); - } finally { resume(); await Promise.allSettled(pending); await service.close(); } + assert.equal((await other).reachable, true); + await Promise.all(pending); + } finally { + resume(); + await Promise.allSettled(pending); + await service.close(); + } +}); + +test('evicts the oldest lease at the global backstop without denying another session', async () => { + const service = new ManagedArtifactPreview(); + try { + const endpoints = []; + for (let index = 0; index < 64; index += 1) { + endpoints.push(await service.prepare('h', client(`preview-${index}`), `s${index}`, 'a1')); + } + const replacement = await service.prepare('h', client('replacement'), 's64', 'a1'); + await assert.rejects(fetch(endpoints[0]!.url)); + assert.equal(await (await fetch(replacement.url)).text(), 'replacement'); + } finally { + await service.close(); + } +}); + +test('releases every preview for a purged session', async () => { + const service = new ManagedArtifactPreview(); + try { + const first = await service.prepare('h', client(), 's1', 'a1'); + const second = await service.prepare('h', client('second'), 's1', 'a2'); + const otherScope = await service.prepare('other-host', client('other scope'), 's1', 'a1'); + await service.releaseSession('h', 's1'); + await assert.rejects(fetch(first.url)); + await assert.rejects(fetch(second.url)); + assert.equal(await (await fetch(otherScope.url)).text(), 'other scope'); + assert.equal((await service.prepare('h', client(), 's1', 'a1')).reachable, true); + } finally { await service.close(); } +}); + +test('delete and Session purge cancel previews that are still preparing', async () => { + for (const release of [ + (service: ManagedArtifactPreview) => service.revoke('h', 's1', 'a1'), + (service: ManagedArtifactPreview) => service.releaseSession('h', 's1'), + ]) { + const service = new ManagedArtifactPreview(); + let resume!: () => void; + const gate = new Promise((resolve) => { resume = resolve; }); + const source = client(); + const preparing = service.prepare('h', { + ...source, + getArtifact: async (...args) => { + await gate; + return source.getArtifact(...args); + }, + }, 's1', 'a1'); + try { + await release(service); + resume(); + await assert.rejects(preparing, /closed/); + } finally { + resume(); + await Promise.allSettled([preparing]); + await service.close(); + } + } }); test('tool binds to the admitted session and returns endpoint evidence only', async () => { diff --git a/apps/desktop/src/main/managed-artifact-preview.ts b/apps/desktop/src/main/managed-artifact-preview.ts index 16d7bf5b75..622a0b5398 100644 --- a/apps/desktop/src/main/managed-artifact-preview.ts +++ b/apps/desktop/src/main/managed-artifact-preview.ts @@ -24,6 +24,10 @@ import type { DesktopRuntimeHostClient } from './runtime-host-client.js'; export const PREVIEW_MAX_BYTES = 8 * 1024 * 1024; const MAX_PREVIEWS = 16; +// Per-Session admission alone cannot bound the Desktop, so one global backstop +// remains. It evicts the oldest lease instead of rejecting the newest: a global +// rejection is what let one Session starve every other for a full TTL. +const MAX_TOTAL_PREVIEWS = 64; const PREVIEW_TTL_MS = 30 * 60 * 1000; const READ_DEADLINE_MS = 30_000; @@ -68,7 +72,16 @@ export class ManagedArtifactPreview { throw new Error('Invalid Artifact identity'); } if (this.closed || this.retiredScopes.has(scope)) throw new Error('Preview owner is closed'); - if (this.leases.size >= MAX_PREVIEWS) throw new Error('Too many active previews; wait for expiry'); + const sessionLeases = [...this.leases].filter( + (lease) => lease.scope === scope && lease.sessionId === sessionId, + ); + if (sessionLeases.length >= MAX_PREVIEWS) { + throw new Error('Too many active previews; wait for expiry'); + } + if (this.leases.size >= MAX_TOTAL_PREVIEWS) { + const oldest = this.leases.values().next().value as Lease | undefined; + if (oldest) void this.release(oldest); + } signal?.throwIfAborted(); // Reserve before asynchronous reads, so concurrent preparations cannot exceed the bound. const lease: Lease = { scope, sessionId, artifactId, server: createServer() }; @@ -156,6 +169,14 @@ export class ManagedArtifactPreview { await Promise.all([...this.leases].filter((lease) => lease.scope === scope && lease.sessionId === sessionId && lease.artifactId === artifactId).map((lease) => this.release(lease))); } + async releaseSession(scope: string, sessionId: string): Promise { + await Promise.all( + [...this.leases] + .filter((lease) => lease.scope === scope && lease.sessionId === sessionId) + .map((lease) => this.release(lease)), + ); + } + async closeScope(scope: string): Promise { this.retiredScopes.add(scope); await Promise.all([...this.leases].filter((lease) => lease.scope === scope).map((lease) => this.release(lease))); diff --git a/apps/desktop/src/main/runtime-host-artifacts-ipc-main.ts b/apps/desktop/src/main/runtime-host-artifacts-ipc-main.ts index 88a7b14e90..ba79ab8a39 100644 --- a/apps/desktop/src/main/runtime-host-artifacts-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-artifacts-ipc-main.ts @@ -83,6 +83,8 @@ export function registerRuntimeHostArtifactsIpc( "artifacts:delete", async (_event, sessionId: string, artifactId: string) => { const result = await deps.client.deleteArtifact(sessionId, artifactId); + // Keep the direct revoke: the Host also publishes artifact.changed, but + // stopping the bytes here must not depend on feed delivery to this Client. await deps.preview?.service.revoke(deps.preview.scope, sessionId, artifactId); return result; }, diff --git a/apps/desktop/src/main/runtime-host-boot.ts b/apps/desktop/src/main/runtime-host-boot.ts index 49386368df..c419591337 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -1559,6 +1559,13 @@ function registerHostClientIpc( const unsubscribeSessionCatalogChanges = client.subscribeSessionCatalogChanges( ({ sessionId }) => emitTargetSessionsChanged("updated", sessionId), ); + const unsubscribeArtifactChanges = client.subscribeArtifactChanges((frame) => { + if (frame.reason === 'deleted') { + void managedArtifactPreview.revoke(scope.targetEpoch, frame.sessionId, frame.artifactId); + } else { + void managedArtifactPreview.releaseSession(scope.targetEpoch, frame.sessionId); + } + }); const unsubscribeProjectCatalogChanges = client.subscribeProjectCatalogChanges(() => { sendToRenderer("projects:changed"); }); @@ -1836,6 +1843,7 @@ function registerHostClientIpc( await managedArtifactPreview.closeScope(scope.targetEpoch); unsubscribeConnectionCatalogChanges(); unsubscribeSessionCatalogChanges(); + unsubscribeArtifactChanges(); unsubscribeProjectCatalogChanges(); unsubscribeScheduledTaskChanges(); runtimePolicyTargets.delete(target); diff --git a/apps/desktop/src/main/runtime-host-client.ts b/apps/desktop/src/main/runtime-host-client.ts index 9f10d1772e..f10b50e4d1 100644 --- a/apps/desktop/src/main/runtime-host-client.ts +++ b/apps/desktop/src/main/runtime-host-client.ts @@ -107,6 +107,7 @@ import { type QueueEntryUpdateInput, type QueueMutationResult, SESSION_TRANSCRIPT_BOOTSTRAP_MAX_BYTES, + type ArtifactChangedFrame, type SessionCatalogChangedFrame, type ScheduledTaskChangedFrame, type SessionCatalogItem, @@ -387,6 +388,11 @@ export class DesktopRuntimeHostClient { return this.connection.subscribeConfigurationChanges(listener); } + subscribeArtifactChanges(listener: (frame: ArtifactChangedFrame) => void): () => void { + this.#assertOpen(); + return this.connection.subscribeArtifactChanges(listener); + } + subscribeConnectionCatalogChanges(listener: (revision: number) => void): () => void { this.#assertOpen(); return this.connection.subscribeConnectionCatalogChanges(listener); diff --git a/packages/cli/src/__tests__/acp-stdio-server.test.ts b/packages/cli/src/__tests__/acp-stdio-server.test.ts index 109b064b10..a6b60beafb 100644 --- a/packages/cli/src/__tests__/acp-stdio-server.test.ts +++ b/packages/cli/src/__tests__/acp-stdio-server.test.ts @@ -704,6 +704,7 @@ describe('Maka ACP stdio server', () => { return subscription; }, subscribeConfigurationChanges: () => () => undefined, + subscribeArtifactChanges: () => () => undefined, subscribeConnectionCatalogChanges: () => () => undefined, subscribeProjectCatalogChanges: () => () => undefined, subscribeSessionCatalogChanges: () => () => undefined, diff --git a/packages/cli/src/__tests__/runtime-host-cli-context.test.ts b/packages/cli/src/__tests__/runtime-host-cli-context.test.ts index 7f91046998..845ec96797 100644 --- a/packages/cli/src/__tests__/runtime-host-cli-context.test.ts +++ b/packages/cli/src/__tests__/runtime-host-cli-context.test.ts @@ -61,6 +61,7 @@ test('CLI Runtime Host bootstrap launches the execution composition', async () = closed: new Promise(() => {}), status: async () => ({ state: 'ready' }), subscribeConfigurationChanges: () => () => {}, + subscribeArtifactChanges: () => () => {}, subscribeConnectionCatalogChanges: () => () => {}, subscribeProjectCatalogChanges: () => () => {}, subscribeSessionCatalogChanges: () => () => {}, @@ -109,6 +110,7 @@ test('connection-only CLI bootstrap does not read the model connection catalog', closed: new Promise(() => {}), status: async () => ({ state: 'ready' }), subscribeConfigurationChanges: () => () => {}, + subscribeArtifactChanges: () => () => {}, subscribeConnectionCatalogChanges: () => () => {}, subscribeProjectCatalogChanges: () => () => {}, subscribeSessionCatalogChanges: () => () => {}, @@ -140,6 +142,7 @@ for (const temporary of [true, false]) { closed: new Promise(() => {}), status: async () => ({ state: 'ready' }), subscribeConfigurationChanges: () => () => {}, + subscribeArtifactChanges: () => () => {}, subscribeConnectionCatalogChanges: () => () => {}, subscribeProjectCatalogChanges: () => () => {}, subscribeSessionCatalogChanges: () => () => {}, @@ -210,6 +213,7 @@ test('CLI Runtime Host bootstrap aborts a stalled catalog read and closes its co closed: new Promise(() => {}), status: async () => ({ state: 'ready' }), subscribeConfigurationChanges: () => () => {}, + subscribeArtifactChanges: () => () => {}, subscribeConnectionCatalogChanges: () => () => {}, subscribeProjectCatalogChanges: () => () => {}, subscribeSessionCatalogChanges: () => () => {}, @@ -378,6 +382,7 @@ test('remote CLI profiles pin root identity and resolve credential outside the p closed: new Promise(() => {}), status: async () => ({ state: 'ready' }), subscribeConfigurationChanges: () => () => {}, + subscribeArtifactChanges: () => () => {}, subscribeConnectionCatalogChanges: () => () => {}, subscribeProjectCatalogChanges: () => () => {}, subscribeSessionCatalogChanges: () => () => {}, @@ -485,6 +490,7 @@ test('remote CLI profile state and Client identity use the explicit Client Data closed: new Promise(() => {}), status: async () => ({ state: 'ready' }), subscribeConfigurationChanges: () => () => {}, + subscribeArtifactChanges: () => () => {}, subscribeConnectionCatalogChanges: () => () => {}, subscribeProjectCatalogChanges: () => () => {}, subscribeSessionCatalogChanges: () => () => {}, @@ -564,6 +570,7 @@ test('remote CLI enables SSH prompts only for an explicitly interactive TTY', as closed: new Promise(() => {}), status: async () => ({ state: 'ready' }), subscribeConfigurationChanges: () => () => {}, + subscribeArtifactChanges: () => () => {}, subscribeConnectionCatalogChanges: () => () => {}, subscribeProjectCatalogChanges: () => () => {}, subscribeSessionCatalogChanges: () => () => {}, @@ -736,6 +743,7 @@ test('local CLI delegates a managed cold start once and reconnects without a lau closed: new Promise(() => {}), close: async () => {}, subscribeConfigurationChanges: () => () => {}, + subscribeArtifactChanges: () => () => {}, subscribeConnectionCatalogChanges: () => () => {}, subscribeProjectCatalogChanges: () => () => {}, subscribeSessionCatalogChanges: () => () => {}, @@ -843,6 +851,7 @@ for (const action of ['cancel', 'interrupt', 'retry'] as const) { closed: new Promise(() => {}), status: async () => ({ state: 'ready' }), subscribeConfigurationChanges: () => () => {}, + subscribeArtifactChanges: () => () => {}, subscribeConnectionCatalogChanges: () => () => {}, subscribeProjectCatalogChanges: () => () => {}, subscribeSessionCatalogChanges: () => () => {}, diff --git a/packages/cli/src/__tests__/runtime-host-onboarding.test.ts b/packages/cli/src/__tests__/runtime-host-onboarding.test.ts index 1f125c96dc..cade2162c4 100644 --- a/packages/cli/src/__tests__/runtime-host-onboarding.test.ts +++ b/packages/cli/src/__tests__/runtime-host-onboarding.test.ts @@ -163,6 +163,7 @@ function oauthPhysicalConnection( return request(operation, input); }, subscribeConfigurationChanges: () => () => {}, + subscribeArtifactChanges: () => () => {}, subscribeConnectionCatalogChanges: () => () => {}, subscribeProjectCatalogChanges: () => () => {}, subscribeSessionCatalogChanges: () => () => {}, diff --git a/packages/cli/src/__tests__/tui-mcp-remote-publication.test.ts b/packages/cli/src/__tests__/tui-mcp-remote-publication.test.ts index f5e9aaf575..1c8c3170b9 100644 --- a/packages/cli/src/__tests__/tui-mcp-remote-publication.test.ts +++ b/packages/cli/src/__tests__/tui-mcp-remote-publication.test.ts @@ -882,6 +882,7 @@ function connectionHarness( return { registrationId: 'registration-a', revision: harness.unregisters }; }, subscribeConfigurationChanges: () => () => undefined, + subscribeArtifactChanges: () => () => undefined, subscribeConnectionCatalogChanges: () => () => undefined, subscribeProjectCatalogChanges: () => () => undefined, subscribeSessionCatalogChanges: () => () => undefined, diff --git a/packages/runtime-host/src/__tests__/artifact-coordinator.test.ts b/packages/runtime-host/src/__tests__/artifact-coordinator.test.ts index 51f2f5e4a7..5c137cdba0 100644 --- a/packages/runtime-host/src/__tests__/artifact-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/artifact-coordinator.test.ts @@ -292,6 +292,7 @@ test('Artifact mutation failure requests Host drain and fails closed', async () store.close(); let drainRequests = 0; + const deletedArtifacts: Array<{ sessionId: string; artifactId: string }> = []; const coordinator = new HostArtifactCoordinator( store, () => { @@ -299,6 +300,9 @@ test('Artifact mutation failure requests Host drain and fails closed', async () }, new SessionAdmissionGate(), { probeSessionRemoval: async () => ({ kind: 'present' }) }, + Date.now, + undefined, + (sessionId, artifactId) => deletedArtifacts.push({ sessionId, artifactId }), ); assert.deepEqual( @@ -318,6 +322,7 @@ test('Artifact mutation failure requests Host drain and fails closed', async () }, ); assert.equal(drainRequests, 1); + assert.deepEqual(deletedArtifacts, []); assert.deepEqual( await coordinator.handlers['artifact.ingest']( { @@ -346,6 +351,70 @@ test('Artifact mutation failure requests Host drain and fails closed', async () } }); +test('Artifact deletion publishes invalidation only after a committed delete', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-artifact-delete-change-')); + const capability = await resolveStorageRoot({ path: root, kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + try { + const store = await openInteractiveArtifactStoreForWrite(owner.lease); + await store.create({ + id: 'user-artifact', + sessionId: 'session-1', + turnId: 'turn-1', + name: 'preview.html', + kind: 'html', + content: 'Preview', + source: 'tool_result', + now: 1, + }); + await store.create({ + id: 'protected-artifact', + sessionId: 'session-1', + turnId: 'turn-1', + name: 'evidence.txt', + kind: 'file', + content: 'evidence', + source: 'deep_research', + now: 2, + }); + const deletedArtifacts: Array<{ sessionId: string; artifactId: string }> = []; + const coordinator = new HostArtifactCoordinator( + store, + () => assert.fail('handled deletion outcomes must not request Host drain'), + new SessionAdmissionGate(), + { probeSessionRemoval: async () => ({ kind: 'present' }) }, + Date.now, + undefined, + (sessionId, artifactId) => deletedArtifacts.push({ sessionId, artifactId }), + ); + + assert.equal( + ( + await coordinator.handlers['artifact.delete']( + { sessionId: 'session-1', artifactId: 'protected-artifact' }, + connectionContext, + ) + ).ok, + false, + ); + assert.deepEqual(deletedArtifacts, []); + + assert.deepEqual( + await coordinator.handlers['artifact.delete']( + { sessionId: 'session-1', artifactId: 'user-artifact' }, + connectionContext, + ), + { ok: true, result: { kind: 'deleted' } }, + ); + assert.deepEqual(deletedArtifacts, [{ sessionId: 'session-1', artifactId: 'user-artifact' }]); + store.close(); + } finally { + await owner.close(); + await rm(root, { recursive: true, force: true }); + } +}); + test('Artifact query streams complete content in bounded ordered chunks', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-artifact-read-chunks-')); const capability = await resolveStorageRoot({ path: root, kind: 'interactive' }); diff --git a/packages/runtime-host/src/__tests__/artifact-protocol.test.ts b/packages/runtime-host/src/__tests__/artifact-protocol.test.ts index 906289e27f..993cffa9e9 100644 --- a/packages/runtime-host/src/__tests__/artifact-protocol.test.ts +++ b/packages/runtime-host/src/__tests__/artifact-protocol.test.ts @@ -41,6 +41,65 @@ import { encodeArtifactProjection } from '../protocol/artifact.js'; const revision = `sha256:${'a'.repeat(64)}` as const; describe('Artifact protocol', () => { + test('accepts closed Artifact change frames and rejects malformed invalidations', () => { + assert.deepEqual( + decodeHostFrame({ + kind: 'artifact.changed', + reason: 'deleted', + sessionId: 'session-1', + artifactId: 'artifact-1', + }), + { + kind: 'artifact.changed', + reason: 'deleted', + sessionId: 'session-1', + artifactId: 'artifact-1', + }, + ); + assert.deepEqual( + decodeHostFrame({ + kind: 'artifact.changed', + reason: 'session_purged', + sessionId: 'session-1', + }), + { + kind: 'artifact.changed', + reason: 'session_purged', + sessionId: 'session-1', + }, + ); + + for (const frame of [ + null, + undefined, + { + kind: 'artifact.changed', + reason: 'deleted', + sessionId: 'session-1', + }, + { + kind: 'artifact.changed', + reason: 'deleted', + sessionId: 'session-1', + artifactId: 'artifact-1', + revision: 1, + }, + { + kind: 'artifact.changed', + reason: 'session_purged', + sessionId: 'session-1', + artifactId: 'artifact-1', + }, + { + kind: 'artifact.changed', + reason: 'unknown', + sessionId: 'session-1', + }, + ]) { + assert.throws(() => decodeHostFrame(frame), isInvalidFrame); + } + }); + test('accepts closed Artifact operations and rejects open shapes', () => { for (const input of [ { kind: 'list_start', sessionId: 'session-1' }, diff --git a/packages/runtime-host/src/__tests__/artifact-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/artifact-two-client-uds.test.ts index 06b45444f4..42665200e2 100644 --- a/packages/runtime-host/src/__tests__/artifact-two-client-uds.test.ts +++ b/packages/runtime-host/src/__tests__/artifact-two-client-uds.test.ts @@ -39,7 +39,11 @@ import { RuntimeHostOperationError, type RuntimeHostConnection, } from '../client/index.js'; -import { RUNTIME_HOST_PROTOCOL_VERSION, type ArtifactQueryResult } from '../protocol/index.js'; +import { + RUNTIME_HOST_PROTOCOL_VERSION, + type ArtifactChangedFrame, + type ArtifactQueryResult, +} from '../protocol/index.js'; import { removePosixEndpointDirectories } from './fixtures/endpoint-hygiene.js'; const CURRENT_PROTOCOL = { @@ -162,12 +166,40 @@ test('production Host ignores Artifact publication residue and preserves deletes ); assert.equal((await getArtifact(desktop, sessionId, deleteA)).artifact?.id, deleteA); + const artifactChanges: ArtifactChangedFrame[] = []; + let resolveArtifactChanges!: () => void; + const receivedArtifactChanges = new Promise((resolve) => { + resolveArtifactChanges = resolve; + }); + const unsubscribeArtifactChanges = desktop.subscribeArtifactChanges((frame) => { + artifactChanges.push(frame); + if (artifactChanges.length === 2) resolveArtifactChanges(); + }); const [deletedA, deletedB] = await Promise.all([ desktop.request('artifact.delete', { sessionId, artifactId: deleteA }), tui.request('artifact.delete', { sessionId, artifactId: deleteB }), ]); assert.deepEqual(deletedA, { kind: 'deleted' }); assert.deepEqual(deletedB, { kind: 'deleted' }); + await withTimeout( + receivedArtifactChanges, + 5_000, + 'Artifact delete invalidations were not delivered', + ); + assert.deepEqual( + artifactChanges + .map((frame) => { + assert.equal(frame.reason, 'deleted'); + return frame.reason === 'deleted' + ? { sessionId: frame.sessionId, artifactId: frame.artifactId } + : undefined; + }) + .sort((left, right) => (left?.artifactId ?? '').localeCompare(right?.artifactId ?? '')), + [ + { sessionId, artifactId: deleteB }, + { sessionId, artifactId: deleteA }, + ].sort((left, right) => left.artifactId.localeCompare(right.artifactId)), + ); await assert.rejects( desktop.request('artifact.delete', { sessionId, artifactId: deleteA }), operationError('not_found'), @@ -187,6 +219,9 @@ test('production Host ignores Artifact publication residue and preserves deletes operationError('operation_conflict'), ); } + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(artifactChanges.length, 2); + unsubscribeArtifactChanges(); const stale = await tui.request('artifact.query', { kind: 'list_continue', diff --git a/packages/runtime-host/src/__tests__/connection-session.test.ts b/packages/runtime-host/src/__tests__/connection-session.test.ts index 62baf17e59..5b3b029ef1 100644 --- a/packages/runtime-host/src/__tests__/connection-session.test.ts +++ b/packages/runtime-host/src/__tests__/connection-session.test.ts @@ -57,6 +57,7 @@ import type { ClientCapabilityService, } from '../server/client-capability-service.js'; import { RuntimeHostConnectionSession } from '../server/connection-session.js'; +import { HostChangeFeed } from '../server/host-change-feed.js'; import type { SessionContinuityService } from '../server/session-continuity-service.js'; import { createUnavailableHostCoreOperationHandlers, @@ -136,6 +137,65 @@ test('concurrent responses remain framed and correlated in reverse completion or ); }); +test('scopes Session Guest Artifact changes to the shared Session', async () => { + const pair = await openTransportPair(); + const hostChanges = new HostChangeFeed(); + const session = new RuntimeHostConnectionSession({ + transport: pair.serverTransport, + connection: { + ...acceptedConnection('session-guest-artifact-changes'), + authority: { + principalKind: 'session_guest', + principalId: 'session_guest:guest-1', + credentialId: 'credential-1', + operationGrants: ['artifact.query', 'session.shared.query'], + canPublishClientCapabilities: false, + canUseHostPaths: false, + }, + }, + resolveHandlers: () => ({ + 'host.status': async () => ({ + ok: true, + result: { + hostEpoch: 'host-epoch', + compositionId: 'maka.interactive', + compositionRevision: '1', + state: 'ready', + connections: 1, + activeOperations: 0, + activeResidencies: 0, + }, + }), + ...UNUSED_HOST_DIAGNOSTICS_HANDLER, + ...createUnavailableHostCoreOperationHandlers(), + ...createUnavailableDomainOperationHandlers(), + }), + resolveContinuity: () => undefined, + resolveHostChanges: () => hostChanges, + resolveSharedSessionId: () => 'session-1', + beginOperation: async () => ({ + acquireResidency: () => ({ release() {} }), + seal() {}, + finish() {}, + }), + onTeardown() {}, + }); + const run = session.run(); + try { + hostChanges.publishArtifactDeleted('session-2', 'artifact-2'); + hostChanges.publishArtifactDeleted('session-1', 'artifact-1'); + assert.deepEqual(decodeHostFrame(await pair.clientTransport.read(1_000)), { + kind: 'artifact.changed', + reason: 'deleted', + sessionId: 'session-1', + artifactId: 'artifact-1', + }); + } finally { + pair.clientTransport.abort(); + await Promise.allSettled([run, pair.close()]); + } +}); + test('transcript pages are serialized per connection before their responses are retained', async () => { const pair = await openTransportPair(); const entered = Array.from({ length: 3 }, () => deferred()); diff --git a/packages/runtime-host/src/__tests__/host-change-feed.test.ts b/packages/runtime-host/src/__tests__/host-change-feed.test.ts index 837b0aad3c..0ba2cde361 100644 --- a/packages/runtime-host/src/__tests__/host-change-feed.test.ts +++ b/packages/runtime-host/src/__tests__/host-change-feed.test.ts @@ -23,11 +23,23 @@ import { HostChangeFeed } from '../server/host-change-feed.js'; test('routes each change kind only to subscribed connections', () => { const feed = new HostChangeFeed(); + const artifact: unknown[] = []; + const scopedArtifact: unknown[] = []; const configuration: unknown[] = []; const project: unknown[] = []; const scopedSession: unknown[] = []; const otherGuest: unknown[] = []; const all: unknown[] = []; + feed.attachConnection( + 'artifact', + { artifact: true }, + { send: async (frame) => void artifact.push(frame) }, + ); + feed.attachConnection( + 'scoped-artifact', + { artifact: { sessionId: 'session-1' } }, + { send: async (frame) => void scopedArtifact.push(frame) }, + ); feed.attachConnection( 'configuration', { configuration: true }, @@ -41,6 +53,7 @@ test('routes each change kind only to subscribed connections', () => { feed.attachConnection( 'all', { + artifact: true, configuration: true, connectionCatalog: true, projectCatalog: true, @@ -60,6 +73,8 @@ test('routes each change kind only to subscribed connections', () => { { send: async (frame) => void otherGuest.push(frame) }, ); + feed.publishArtifactDeleted('session-1', 'artifact-1'); + feed.publishArtifactSessionPurged('session-2'); feed.publishConfiguration(); feed.publishConnectionCatalog(); feed.publishProjectCatalog(); @@ -69,6 +84,27 @@ test('routes each change kind only to subscribed connections', () => { feed.publishSessionCatalog('session-1'); feed.publishScheduledTask(7, 'updated', 'task-1'); + assert.deepEqual(artifact, [ + { + kind: 'artifact.changed', + reason: 'deleted', + sessionId: 'session-1', + artifactId: 'artifact-1', + }, + { + kind: 'artifact.changed', + reason: 'session_purged', + sessionId: 'session-2', + }, + ]); + assert.deepEqual(scopedArtifact, [ + { + kind: 'artifact.changed', + reason: 'deleted', + sessionId: 'session-1', + artifactId: 'artifact-1', + }, + ]); assert.deepEqual( configuration.map((frame) => (frame as { kind: string }).kind), ['configuration.changed'], @@ -78,7 +114,7 @@ test('routes each change kind only to subscribed connections', () => { project.map((frame) => (frame as { kind: string }).kind), ['project.catalog.changed'], ); - assert.equal(all.length, 8); + assert.equal(all.length, 10); assert.deepEqual(scopedSession, [ { kind: 'session.catalog.changed', revision: 1, sessionId: 'session-1' }, { kind: 'session.catalog.changed', revision: 3, sessionId: 'session-1' }, diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index 63c9a40b92..d00eaf65ce 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -134,6 +134,10 @@ describe('Runtime Host bootstrap protocol', () => { assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 22); }); + test('publishes a new compatibility epoch for Artifact invalidation frames', () => { + assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 161); + }); + test('publishes a new compatibility epoch for mandatory submit Skill outcomes', () => { // Submit Skill outcomes and explicit OAuth Connection targets independently // claimed epoch 78, so their merge requires a distinct compatibility boundary. diff --git a/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts b/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts index 337cd01a71..337e1b9697 100644 --- a/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts +++ b/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts @@ -36,6 +36,7 @@ import { INTERACTIVE_RUNTIME_HOST_COMPOSITION_ID, RUNTIME_HOST_COMPATIBILITY_EPOCH, RUNTIME_HOST_PROTOCOL_VERSION, + type ArtifactChangedFrame, type HostIncompatible, type HostStatusResult, type OperationInput, @@ -128,6 +129,68 @@ test('a reconnecting Client reports the connection generation used by direct ope await connection.close(); }); +test('a reconnecting Client forwards Artifact invalidations only from its current connection', async () => { + const first = connectionHarness('first', () => undefined); + const replacement = connectionHarness('replacement', () => undefined); + const connection = await createRuntimeHostReconnectingConnection({ + initialConnection: first.connection, + connect: async () => replacement.connection, + }); + const changes: ArtifactChangedFrame[] = []; + const unsubscribe = connection.subscribeArtifactChanges((frame) => changes.push(frame)); + + first.emitArtifactChange({ + kind: 'artifact.changed', + reason: 'deleted', + sessionId: 'session-1', + artifactId: 'artifact-1', + }); + assert.deepEqual(changes, [ + { + kind: 'artifact.changed', + reason: 'deleted', + sessionId: 'session-1', + artifactId: 'artifact-1', + }, + ]); + + first.disconnect(); + await waitForCondition(() => replacement.artifactSubscribers === 1); + first.emitArtifactChange({ + kind: 'artifact.changed', + reason: 'session_purged', + sessionId: 'stale-session', + }); + replacement.emitArtifactChange({ + kind: 'artifact.changed', + reason: 'session_purged', + sessionId: 'session-2', + }); + assert.deepEqual(changes, [ + { + kind: 'artifact.changed', + reason: 'deleted', + sessionId: 'session-1', + artifactId: 'artifact-1', + }, + { + kind: 'artifact.changed', + reason: 'session_purged', + sessionId: 'session-2', + }, + ]); + + unsubscribe(); + replacement.emitArtifactChange({ + kind: 'artifact.changed', + reason: 'deleted', + sessionId: 'session-2', + artifactId: 'artifact-2', + }); + assert.equal(changes.length, 2); + await connection.close(); +}); + test('a reconnecting Client retries status through the validated status surface', async () => { let firstStatusCalls = 0; let replacementStatusCalls = 0; @@ -874,6 +937,7 @@ function connectionHarness( resolveClosed = resolve; }); const operations: DirectRequestOperationKey[] = []; + const artifactListeners = new Set<(frame: ArtifactChangedFrame) => void>(); let openedSubscriptions = 0; const connection = { rootId: 'root-id', @@ -893,6 +957,10 @@ function connectionHarness( return openSubscription(); }, subscribeConfigurationChanges: () => () => {}, + subscribeArtifactChanges: (listener: (frame: ArtifactChangedFrame) => void) => { + artifactListeners.add(listener); + return () => artifactListeners.delete(listener); + }, subscribeConnectionCatalogChanges: () => () => {}, subscribeProjectCatalogChanges: () => () => {}, subscribeSessionCatalogChanges: () => () => {}, @@ -903,6 +971,12 @@ function connectionHarness( connection, operations, disconnect: resolveClosed, + emitArtifactChange: (frame: ArtifactChangedFrame) => { + for (const listener of artifactListeners) listener(frame); + }, + get artifactSubscribers() { + return artifactListeners.size; + }, get openedSubscriptions() { return openedSubscriptions; }, diff --git a/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts index b24ca8e455..fde5f7e3a8 100644 --- a/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts @@ -196,6 +196,56 @@ describe('Host Session retirement coordinator', () => { assert.equal(garbageBatches, 0); }); + test('publishes Session Artifact invalidation only when Artifact purge succeeds', async () => { + const purged: string[] = []; + await purgeSessionSidecars( + { + artifacts: { purgeSessionArtifacts: async () => {} }, + sessionTodo: { purgeSessionState: async () => {} }, + purgeOperationalState: async () => {}, + onArtifactsPurged: (sessionId) => purged.push(sessionId), + }, + 'session-success', + ); + assert.deepEqual(purged, ['session-success']); + + await assert.rejects( + purgeSessionSidecars( + { + artifacts: { + purgeSessionArtifacts: async () => { + throw new Error('Artifact purge failed'); + }, + }, + sessionTodo: { purgeSessionState: async () => {} }, + purgeOperationalState: async () => {}, + onArtifactsPurged: (sessionId) => purged.push(sessionId), + }, + 'session-artifact-failure', + ), + AggregateError, + ); + assert.deepEqual(purged, ['session-success']); + + await assert.rejects( + purgeSessionSidecars( + { + artifacts: { purgeSessionArtifacts: async () => {} }, + sessionTodo: { + purgeSessionState: async () => { + throw new Error('Todo purge failed'); + }, + }, + purgeOperationalState: async () => {}, + onArtifactsPurged: (sessionId) => purged.push(sessionId), + }, + 'session-other-failure', + ), + AggregateError, + ); + assert.deepEqual(purged, ['session-success', 'session-other-failure']); + }); + test('rejects ordinary archive and remove operations for the Coordination Session', async () => { await withHarness(async (harness) => { const created = await harness.store.createStableSession({ diff --git a/packages/runtime-host/src/client/connection.ts b/packages/runtime-host/src/client/connection.ts index aed036233b..8c9dc16af8 100644 --- a/packages/runtime-host/src/client/connection.ts +++ b/packages/runtime-host/src/client/connection.ts @@ -39,6 +39,7 @@ import { type ClientCapabilityReplaceResult, type ClientCapabilityUnregisterResult, type ClientHello, + type ArtifactChangedFrame, type ConfigurationChangedFrame, type ConnectionCatalogChangedFrame, type HostOperationErrorCode, @@ -286,6 +287,7 @@ export interface RuntimeHostConnection { options?: number | ClientCapabilityRegistrationOptions, ): Promise; subscribeConfigurationChanges(listener: (revision: number) => void): () => void; + subscribeArtifactChanges(listener: (frame: ArtifactChangedFrame) => void): () => void; subscribeConnectionCatalogChanges(listener: (revision: number) => void): () => void; subscribeProjectCatalogChanges(listener: (revision: number) => void): () => void; subscribeSessionCatalogChanges(listener: (frame: SessionCatalogChangedFrame) => void): () => void; @@ -388,6 +390,7 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { readonly #retiredSubscriptionIds = new Set(); readonly #clientCapabilities: ClientCapabilityChannel; readonly #configurationChangeListeners = new Set<(revision: number) => void>(); + readonly #artifactChangeListeners = new Set<(frame: ArtifactChangedFrame) => void>(); readonly #connectionCatalogChangeListeners = new Set<(revision: number) => void>(); readonly #projectCatalogChangeListeners = new Set<(revision: number) => void>(); readonly #sessionCatalogChangeListeners = new Set<(frame: SessionCatalogChangedFrame) => void>(); @@ -723,6 +726,11 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { return () => this.#configurationChangeListeners.delete(listener); } + subscribeArtifactChanges(listener: (frame: ArtifactChangedFrame) => void): () => void { + this.#artifactChangeListeners.add(listener); + return () => this.#artifactChangeListeners.delete(listener); + } + subscribeConnectionCatalogChanges(listener: (revision: number) => void): () => void { this.#connectionCatalogChangeListeners.add(listener); return () => this.#connectionCatalogChangeListeners.delete(listener); @@ -756,6 +764,9 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { continue; } switch (frame.kind) { + case 'artifact.changed': + this.#acceptArtifactChanged(frame); + continue; case 'configuration.changed': this.#acceptConfigurationChanged(frame); continue; @@ -838,6 +849,16 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { } } + #acceptArtifactChanged(frame: ArtifactChangedFrame): void { + for (const listener of this.#artifactChangeListeners) { + try { + listener(frame); + } catch { + // A presentation listener cannot invalidate the Host connection. + } + } + } + #acceptConnectionCatalogChanged(frame: ConnectionCatalogChangedFrame): void { for (const listener of this.#connectionCatalogChangeListeners) { try { @@ -1043,6 +1064,7 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { this.#retiredSubscriptionIds.clear(); this.#clientCapabilities.close(error); this.#configurationChangeListeners.clear(); + this.#artifactChangeListeners.clear(); this.#sessionCatalogChangeListeners.clear(); this.#scheduledTaskChangeListeners.clear(); if (gracefulPeerClose) this.#transport.closeAfterFlush(); diff --git a/packages/runtime-host/src/client/reconnecting-connection.ts b/packages/runtime-host/src/client/reconnecting-connection.ts index 9e2885c7a5..917beed4ac 100644 --- a/packages/runtime-host/src/client/reconnecting-connection.ts +++ b/packages/runtime-host/src/client/reconnecting-connection.ts @@ -25,6 +25,7 @@ import { type OperationInput, type OperationKey, type OperationOutput, + type ArtifactChangedFrame, type SessionCatalogChangedFrame, type ScheduledTaskChangedFrame, type SubscriptionOpenInput, @@ -128,6 +129,7 @@ class RuntimeHostReconnectingConnectionImpl implements RuntimeHostReconnectingCo readonly #connectionCatalogListeners = new Set<(revision: number) => void>(); readonly #projectListeners = new Set<(revision: number) => void>(); readonly #sessionListeners = new Set<(frame: SessionCatalogChangedFrame) => void>(); + readonly #artifactListeners = new Set<(frame: ArtifactChangedFrame) => void>(); readonly #scheduledTaskListeners = new Set<(frame: ScheduledTaskChangedFrame) => void>(); readonly #lifecycle: RuntimeHostReconnectLifecycle; #connectionAvailability: RuntimeHostConnectionAvailability; @@ -246,6 +248,11 @@ class RuntimeHostReconnectingConnectionImpl implements RuntimeHostReconnectingCo return () => this.#configurationListeners.delete(listener); } + subscribeArtifactChanges(listener: (frame: ArtifactChangedFrame) => void): () => void { + this.#artifactListeners.add(listener); + return () => this.#artifactListeners.delete(listener); + } + subscribeConnectionAvailability( listener: (availability: RuntimeHostConnectionAvailability) => void, ): () => void { @@ -359,6 +366,9 @@ class RuntimeHostReconnectingConnectionImpl implements RuntimeHostReconnectingCo connection.subscribeConfigurationChanges((revision: number) => { notify(this.#configurationListeners, revision); }), + connection.subscribeArtifactChanges((frame: ArtifactChangedFrame) => { + notify(this.#artifactListeners, frame); + }), connection.subscribeConnectionCatalogChanges((revision: number) => { notify(this.#connectionCatalogListeners, revision); }), diff --git a/packages/runtime-host/src/protocol/artifact-change.ts b/packages/runtime-host/src/protocol/artifact-change.ts new file mode 100644 index 0000000000..88a19ab106 --- /dev/null +++ b/packages/runtime-host/src/protocol/artifact-change.ts @@ -0,0 +1,62 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { requireExactRecord, requireId, requireRecord } from './codec.js'; +import { invalidProtocolFrame } from './errors.js'; + +export type ArtifactChangedFrame = + | { + readonly kind: 'artifact.changed'; + readonly reason: 'deleted'; + readonly sessionId: string; + readonly artifactId: string; + } + | { + readonly kind: 'artifact.changed'; + readonly reason: 'session_purged'; + readonly sessionId: string; + }; + +export function decodeArtifactChangedFrame(value: unknown): ArtifactChangedFrame { + const candidate = requireRecord(value, 'Artifact changed frame'); + const fields = + candidate.reason === 'deleted' + ? ['kind', 'reason', 'sessionId', 'artifactId'] + : ['kind', 'reason', 'sessionId']; + const frame = requireExactRecord(value, 'Artifact changed frame', fields); + if (frame.kind !== 'artifact.changed') { + throw invalidProtocolFrame('Invalid Artifact changed frame kind'); + } + if (frame.reason === 'deleted') { + return { + kind: 'artifact.changed', + reason: 'deleted', + sessionId: requireId(frame.sessionId, 'sessionId'), + artifactId: requireId(frame.artifactId, 'artifactId'), + }; + } + if (frame.reason !== 'session_purged') { + throw invalidProtocolFrame('Invalid Artifact change reason'); + } + return { + kind: 'artifact.changed', + reason: 'session_purged', + sessionId: requireId(frame.sessionId, 'sessionId'), + }; +} diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 2a4cd0b470..40f7e1bb4d 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -41,6 +41,7 @@ import { decodeConfigurationChangedFrame, type ConfigurationChangedFrame, } from './configuration-change.js'; +import { decodeArtifactChangedFrame, type ArtifactChangedFrame } from './artifact-change.js'; import { decodeSessionCatalogChangedFrame, type SessionCatalogChangedFrame, @@ -68,6 +69,7 @@ import { isCanonicalRuntimeHostWebSocketPath } from './websocket-path.js'; import { INTERACTIVE_RUNTIME_HOST_COMPOSITION_ID } from '../composition-identity.js'; export * from './access-authority.js'; +export * from './artifact-change.js'; export * from './agent-graph.js'; export * from './interaction.js'; export * from './daily-review.js'; @@ -103,7 +105,8 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 183 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 184 as const; +// 184: Host change feed includes Artifact deletion and Session purge invalidation frames. // 183: Jev policy snapshots, set_jev mutation and credential locator require matching peers. // 182: Executor catalogs expose structured model families and thinking variant IDs. // 181: Canonical executor models and retained provider stop reasons after cancellation. @@ -533,6 +536,7 @@ export type HostFrame = | SubscriptionFrame | ClientCapabilityHostFrame | ConfigurationChangedFrame + | ArtifactChangedFrame | ConnectionCatalogChangedFrame | ProjectCatalogChangedFrame | SessionCatalogChangedFrame @@ -667,6 +671,7 @@ export function decodeHostFrame(value: unknown): HostFrame { return decodeClientCapabilityHostFrame(frame); } if (frame.kind === 'configuration.changed') return decodeConfigurationChangedFrame(frame); + if (frame.kind === 'artifact.changed') return decodeArtifactChangedFrame(frame); if (frame.kind === 'connection.catalog.changed') { return decodeConnectionCatalogChangedFrame(frame); } diff --git a/packages/runtime-host/src/server/artifact-coordinator.ts b/packages/runtime-host/src/server/artifact-coordinator.ts index f58585b6b2..78bfe7de68 100644 --- a/packages/runtime-host/src/server/artifact-coordinator.ts +++ b/packages/runtime-host/src/server/artifact-coordinator.ts @@ -79,6 +79,7 @@ export class HostArtifactCoordinator { | Pick | undefined; readonly #uploads: ConnectionBoundChunkUploads; + readonly #onArtifactDeleted: ((sessionId: string, artifactId: string) => void) | undefined; constructor( store: InteractiveArtifactStoreWriter, @@ -87,12 +88,14 @@ export class HostArtifactCoordinator { sessions: SessionPresenceReader, now: () => number = Date.now, sessionAccessAuthority?: Pick, + onArtifactDeleted?: (sessionId: string, artifactId: string) => void, ) { this.#store = authenticateInteractiveArtifactStoreWriter(store); this.#requestDrain = requestDrain; this.#sessionAdmission = sessionAdmission; this.#sessions = sessions; this.#sessionAccessAuthority = sessionAccessAuthority; + this.#onArtifactDeleted = onArtifactDeleted; this.#uploads = new ConnectionBoundChunkUploads( { maxActive: MAX_ACTIVE_ARTIFACT_UPLOADS, @@ -492,6 +495,7 @@ export class HostArtifactCoordinator { }, }; } + this.#onArtifactDeleted?.(input.sessionId, input.artifactId); return { ok: true, result: encodeArtifactDeleteResult({ kind: 'deleted' }), diff --git a/packages/runtime-host/src/server/connection-session.ts b/packages/runtime-host/src/server/connection-session.ts index 7c1de224f7..48537b05b3 100644 --- a/packages/runtime-host/src/server/connection-session.ts +++ b/packages/runtime-host/src/server/connection-session.ts @@ -366,9 +366,18 @@ export class RuntimeHostConnectionSession { : hasRuntimeHostOperationGrant(this.#options.connection.authority, 'session.catalog.query') ? true : undefined; + const artifact: HostChangeSubscriptionMask['artifact'] = + this.#options.connection.authority.principalKind === 'session_guest' + ? sharedSessionId !== undefined + ? { sessionId: sharedSessionId } + : undefined + : hasRuntimeHostOperationGrant(this.#options.connection.authority, 'artifact.query') + ? true + : undefined; this.#hostChanges = service.attachConnection( this.#options.connection.connectionId, { + artifact, configuration: hasRuntimeHostOperationGrant( this.#options.connection.authority, 'runtime.policy.query', diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 861678ba43..ea0476d6e4 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -1657,6 +1657,7 @@ export async function createExecutionRuntimeHostComposition( stores.sessionStore, Date.now, context.sessionAccessAuthority, + (sessionId, artifactId) => hostChanges.publishArtifactDeleted(sessionId, artifactId), ); rootCoordinator = new RootTurnCoordinator( manager, @@ -2666,6 +2667,7 @@ export async function createExecutionRuntimeHostComposition( graph: requireGraphCoordinator(graphCoordinator), isSessionActive: (sessionId) => coordinator.readRootState(sessionId).kind !== 'idle', requestDrain: context.requestDrain, + onArtifactsPurged: (sessionId) => hostChanges.publishArtifactSessionPurged(sessionId), }); const sessionRetirement = new HostSessionRetirementCoordinator({ stores: stores.sessionStore, @@ -2699,6 +2701,7 @@ export async function createExecutionRuntimeHostComposition( // being removed would only wake subscribers to read nothing. await openedPlanStore.purgeSessionState(sessionId); }, + onArtifactsPurged: (sessionId) => hostChanges.publishArtifactSessionPurged(sessionId), purgeAgentGraphState: async (sessionId) => { for (const graphId of await requireGraphCoordinator(graphCoordinator).listGraphIds( sessionId, diff --git a/packages/runtime-host/src/server/host-change-feed.ts b/packages/runtime-host/src/server/host-change-feed.ts index dc5080d5e8..bbcc83b4c6 100644 --- a/packages/runtime-host/src/server/host-change-feed.ts +++ b/packages/runtime-host/src/server/host-change-feed.ts @@ -18,6 +18,7 @@ */ import type { + ArtifactChangedFrame, ConfigurationChangedFrame, ConnectionCatalogChangedFrame, ProjectCatalogChangedFrame, @@ -27,6 +28,7 @@ import type { } from '../protocol/index.js'; export type HostChangeFrame = + | ArtifactChangedFrame | ConfigurationChangedFrame | ConnectionCatalogChangedFrame | ProjectCatalogChangedFrame @@ -38,6 +40,7 @@ export interface HostChangeSubscription { } export interface HostChangeSubscriptionMask { + readonly artifact?: true | { readonly sessionId: string }; readonly configuration?: boolean; readonly connectionCatalog?: boolean; readonly projectCatalog?: boolean; @@ -85,6 +88,14 @@ export class HostChangeFeed { }); } + publishArtifactDeleted(sessionId: string, artifactId: string): void { + this.#publish({ kind: 'artifact.changed', reason: 'deleted', sessionId, artifactId }); + } + + publishArtifactSessionPurged(sessionId: string): void { + this.#publish({ kind: 'artifact.changed', reason: 'session_purged', sessionId }); + } + /** The Host now resolves connection catalogs differently; clients re-read. */ publishConnectionCatalog(): void { this.#connectionCatalogRevision += 1; @@ -119,12 +130,13 @@ export class HostChangeFeed { sessionId, }; for (const [connectionId, subscription] of this.#subscriptions) { - if (!isSubscribed(subscription.mask, frame)) continue; - void subscription.sink.send(frame).catch(() => { - if (this.#subscriptions.get(connectionId) === subscription) { - this.#subscriptions.delete(connectionId); - } - }); + if (isSubscribed(subscription.mask, frame)) { + void subscription.sink.send(frame).catch(() => { + if (this.#subscriptions.get(connectionId) === subscription) { + this.#subscriptions.delete(connectionId); + } + }); + } if ( closeScopeFor !== undefined && subscription.mask.sessionCatalog !== true && @@ -160,6 +172,11 @@ export class HostChangeFeed { function isSubscribed(mask: HostChangeSubscriptionMask, frame: HostChangeFrame): boolean { switch (frame.kind) { + case 'artifact.changed': + return ( + mask.artifact === true || + (mask.artifact !== undefined && frame.sessionId === mask.artifact.sessionId) + ); case 'configuration.changed': return mask.configuration === true; case 'connection.catalog.changed': diff --git a/packages/runtime-host/src/server/session-retirement-coordinator.ts b/packages/runtime-host/src/server/session-retirement-coordinator.ts index e9dc4cb8e8..9ddbb19b61 100644 --- a/packages/runtime-host/src/server/session-retirement-coordinator.ts +++ b/packages/runtime-host/src/server/session-retirement-coordinator.ts @@ -126,6 +126,7 @@ export interface HostSessionRetirementCoordinatorOptions { readonly sessionTodo: Pick; readonly contextOffload?: Pick; readonly purgeOperationalState: (sessionId: string) => Promise; + readonly onArtifactsPurged?: (sessionId: string) => void; readonly purgeAgentGraphState: (sessionId: string) => Promise; readonly worktrees?: Pick; readonly requestDrain: () => void; @@ -198,6 +199,7 @@ export class HostSessionRetirementCoordinator { readonly #sessionTodo: HostSessionRetirementCoordinatorOptions['sessionTodo']; readonly #contextOffload: HostSessionRetirementCoordinatorOptions['contextOffload']; readonly #purgeOperationalState: HostSessionRetirementCoordinatorOptions['purgeOperationalState']; + readonly #onArtifactsPurged: HostSessionRetirementCoordinatorOptions['onArtifactsPurged']; readonly #purgeAgentGraphState: HostSessionRetirementCoordinatorOptions['purgeAgentGraphState']; readonly #worktrees: HostSessionRetirementCoordinatorOptions['worktrees']; readonly #requestDrain: () => void; @@ -226,6 +228,7 @@ export class HostSessionRetirementCoordinator { this.#sessionTodo = options.sessionTodo; this.#contextOffload = options.contextOffload; this.#purgeOperationalState = options.purgeOperationalState; + this.#onArtifactsPurged = options.onArtifactsPurged; this.#purgeAgentGraphState = options.purgeAgentGraphState; this.#worktrees = options.worktrees; this.#requestDrain = options.requestDrain; @@ -733,6 +736,7 @@ export class HostSessionRetirementCoordinator { sessionTodo: this.#sessionTodo, ...(this.#contextOffload ? { contextOffload: this.#contextOffload } : {}), purgeOperationalState: this.#purgeOperationalState, + ...(this.#onArtifactsPurged ? { onArtifactsPurged: this.#onArtifactsPurged } : {}), }, sessionId, ), diff --git a/packages/runtime-host/src/server/session-revision-coordinator.ts b/packages/runtime-host/src/server/session-revision-coordinator.ts index 595494963c..cb6a41af33 100644 --- a/packages/runtime-host/src/server/session-revision-coordinator.ts +++ b/packages/runtime-host/src/server/session-revision-coordinator.ts @@ -122,6 +122,7 @@ export interface HostSessionRevisionCoordinatorOptions { >; readonly isSessionActive: (sessionId: string) => boolean; readonly requestDrain: () => void; + readonly onArtifactsPurged?: (sessionId: string) => void; } /** Host authority for exact, retryable cross-Session branch and revision copies. */ @@ -887,6 +888,9 @@ export class HostSessionRevisionCoordinator { ...(this.options.contextOffload ? { contextOffload: this.options.contextOffload } : {}), purgeOperationalState: (sessionId) => this.#stores.purgeConversationOperationalState(sessionId), + ...(this.options.onArtifactsPurged + ? { onArtifactsPurged: this.options.onArtifactsPurged } + : {}), }, header.id, ); diff --git a/packages/runtime-host/src/server/session-sidecar-purge.ts b/packages/runtime-host/src/server/session-sidecar-purge.ts index 59484fcf12..35d900eff6 100644 --- a/packages/runtime-host/src/server/session-sidecar-purge.ts +++ b/packages/runtime-host/src/server/session-sidecar-purge.ts @@ -26,19 +26,23 @@ export interface SessionSidecarPurgeAuthority { readonly sessionTodo: Pick; readonly contextOffload?: Pick; readonly purgeOperationalState: (sessionId: string) => Promise; + readonly onArtifactsPurged?: (sessionId: string) => void; } export async function purgeSessionSidecars( authority: SessionSidecarPurgeAuthority, sessionId: string, ): Promise { - const outcomes = await Promise.allSettled([ + const [artifactPurge, ...sidecarPurges] = await Promise.allSettled([ authority.artifacts.purgeSessionArtifacts(sessionId), authority.sessionTodo.purgeSessionState(sessionId), ...(authority.contextOffload ? [authority.contextOffload.retireSession(sessionId)] : []), authority.purgeOperationalState(sessionId), ]); - const failures = outcomes.flatMap((outcome) => + // Once Artifacts are purged their previews must stop serving, even when a + // sibling sidecar fails and the aggregate error is thrown below. + if (artifactPurge.status === 'fulfilled') authority.onArtifactsPurged?.(sessionId); + const failures = [artifactPurge, ...sidecarPurges].flatMap((outcome) => outcome.status === 'rejected' ? [outcome.reason] : [], ); if (failures.length > 0) { From de601d234373b2f1a3848fcb7cccdf7d7292ab3f Mon Sep 17 00:00:00 2001 From: SummerC0zyR0ck Date: Thu, 17 Sep 2026 02:59:33 +0000 Subject: [PATCH 2/5] test(desktop): verify previews close on Host disconnect --- .../runtime-host-desktop-candidate.test.ts | 46 +++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts b/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts index 73ad57e8c3..cf60333da5 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts @@ -61,6 +61,7 @@ import { RuntimeHostReconnectingIpcMain } from '../runtime-host-reconnecting-ipc import { desktopSessionResourceKey } from '../../shared/runtime-host-identity.js'; import { waitFor as pollFor } from '@maka/core/test-only/async-primitives'; import { canRepairManagedRuntimeHostStartup } from '../runtime-host-startup-recovery.js'; +import { ManagedArtifactPreview } from '../managed-artifact-preview.js'; const TEST_HOST_ID = 'a'.repeat(64); const TEST_TARGET_EPOCH = 'test-target-epoch'; @@ -499,6 +500,51 @@ test('tears down the whole candidate when the Host connection closes', async () assert.equal(host.closeCalls, 1); }); +test('closes managed Artifact previews when the Host connection closes', async () => { + const ipc = ipcHarness(); + const host = connectionHarness('preview-closed'); + const preview = new ManagedArtifactPreview(); + const bytes = Buffer.from('Preview'); + const candidate = await createDesktopRuntimeHostCandidate(host.connection, { + ...deps(ipc), + registerClientIpc: (_client, _ipc, _controls, _target, scope) => + () => preview.closeScope(scope.targetEpoch), + }); + + try { + const endpoint = await preview.prepare( + TEST_TARGET_EPOCH, + { + getArtifact: async () => ({ + id: 'artifact-1', + sessionId: 'session-1', + turnId: 'turn-1', + createdAt: 0, + name: 'preview.html', + kind: 'html', + sizeBytes: bytes.length, + source: 'tool_result', + }), + streamArtifact: async (_sessionId, _artifactId, write) => { + await write(bytes); + return bytes.length; + }, + }, + 'session-1', + 'artifact-1', + ); + assert.equal(await (await fetch(endpoint.url)).text(), bytes.toString()); + + host.disconnect(); + await candidate.closed; + + await assert.rejects(fetch(endpoint.url)); + } finally { + await candidate.close(); + await preview.close(); + } +}); + test('preserves supported IPC when the connection closes before candidate startup returns', { timeout: 5_000 }, async (t) => { const ipc = ipcHarness(); const router = new RuntimeHostReconnectingIpcMain(ipc); From a86403d11a37b60742634b8232a6ddb6b0f272ff Mon Sep 17 00:00:00 2001 From: SummerC0zyR0ck Date: Thu, 17 Sep 2026 03:33:18 +0000 Subject: [PATCH 3/5] fix(desktop): reopen artifact previews after reconnect --- .../managed-artifact-preview.test.ts | 2 + .../runtime-host-desktop-candidate.test.ts | 66 ++++++++++++------- .../src/main/managed-artifact-preview.ts | 4 ++ apps/desktop/src/main/runtime-host-boot.ts | 1 + 4 files changed, 48 insertions(+), 25 deletions(-) diff --git a/apps/desktop/src/main/__tests__/managed-artifact-preview.test.ts b/apps/desktop/src/main/__tests__/managed-artifact-preview.test.ts index c0481c51bc..65a3fd12ad 100644 --- a/apps/desktop/src/main/__tests__/managed-artifact-preview.test.ts +++ b/apps/desktop/src/main/__tests__/managed-artifact-preview.test.ts @@ -85,6 +85,8 @@ test('isolates leases by origin and rejects credentials for another preview', as await assert.rejects(fetch(first.url)); assert.equal(await (await fetch(second.url)).text(), 'second'); await assert.rejects(service.prepare('host1', client(), 's1', 'a1'), /closed/); + service.openScope('host1'); + assert.equal(await (await fetch((await service.prepare('host1', client(), 's1', 'a1')).url)).text(), html); await service.revoke('host2', 's1', 'a1'); await assert.rejects(fetch(second.url)); } finally { await service.close(); } diff --git a/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts b/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts index cf60333da5..c368386865 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts @@ -500,47 +500,63 @@ test('tears down the whole candidate when the Host connection closes', async () assert.equal(host.closeCalls, 1); }); -test('closes managed Artifact previews when the Host connection closes', async () => { +test('closes old managed Artifact previews and reopens the scope after reconnect', async () => { const ipc = ipcHarness(); - const host = connectionHarness('preview-closed'); + const firstHost = connectionHarness('preview-first'); const preview = new ManagedArtifactPreview(); const bytes = Buffer.from('Preview'); - const candidate = await createDesktopRuntimeHostCandidate(host.connection, { + const candidateDeps = { ...deps(ipc), - registerClientIpc: (_client, _ipc, _controls, _target, scope) => - () => preview.closeScope(scope.targetEpoch), - }); + registerClientIpc: (_client, _ipc, _controls, _target, scope) => { + preview.openScope(scope.targetEpoch); + return () => preview.closeScope(scope.targetEpoch); + }, + } satisfies DesktopRuntimeHostCandidateDeps; + const firstCandidate = await createDesktopRuntimeHostCandidate(firstHost.connection, candidateDeps); + let secondCandidate: Awaited> | undefined; + const source = { + getArtifact: async () => ({ + id: 'artifact-1', + sessionId: 'session-1', + turnId: 'turn-1', + createdAt: 0, + name: 'preview.html', + kind: 'html' as const, + sizeBytes: bytes.length, + source: 'tool_result' as const, + }), + streamArtifact: async (_sessionId: string, _artifactId: string, write: (chunk: Uint8Array) => Promise) => { + await write(bytes); + return bytes.length; + }, + }; try { const endpoint = await preview.prepare( TEST_TARGET_EPOCH, - { - getArtifact: async () => ({ - id: 'artifact-1', - sessionId: 'session-1', - turnId: 'turn-1', - createdAt: 0, - name: 'preview.html', - kind: 'html', - sizeBytes: bytes.length, - source: 'tool_result', - }), - streamArtifact: async (_sessionId, _artifactId, write) => { - await write(bytes); - return bytes.length; - }, - }, + source, 'session-1', 'artifact-1', ); assert.equal(await (await fetch(endpoint.url)).text(), bytes.toString()); - host.disconnect(); - await candidate.closed; + firstHost.disconnect(); + await firstCandidate.closed; await assert.rejects(fetch(endpoint.url)); + + const secondHost = connectionHarness('preview-second'); + secondCandidate = await createDesktopRuntimeHostCandidate(secondHost.connection, candidateDeps); + const replacement = await preview.prepare( + TEST_TARGET_EPOCH, + source, + 'session-1', + 'artifact-1', + ); + assert.equal(await (await fetch(replacement.url)).text(), bytes.toString()); } finally { - await candidate.close(); + await firstCandidate.close(); + await secondCandidate?.close(); await preview.close(); } }); diff --git a/apps/desktop/src/main/managed-artifact-preview.ts b/apps/desktop/src/main/managed-artifact-preview.ts index 622a0b5398..c26d00ce87 100644 --- a/apps/desktop/src/main/managed-artifact-preview.ts +++ b/apps/desktop/src/main/managed-artifact-preview.ts @@ -56,6 +56,10 @@ export class ManagedArtifactPreview { constructor(private readonly ttlMs = PREVIEW_TTL_MS) {} + openScope(scope: string): void { + this.retiredScopes.delete(scope); + } + async releaseUrl(url: string): Promise { const lease = [...this.leases].find((entry) => entry.url === url); if (lease) await this.release(lease); diff --git a/apps/desktop/src/main/runtime-host-boot.ts b/apps/desktop/src/main/runtime-host-boot.ts index c419591337..158a17b240 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -1837,6 +1837,7 @@ function registerHostClientIpc( }); registerOnboardingIpc({ onboardingService, ipcMain: scopedIpc }); registerTaskSubmissionReadinessIpc(taskSubmissionReadinessService, scopedIpc); + managedArtifactPreview.openScope(scope.targetEpoch); return async () => { clientPluginTransport.release(client); unsubscribeConfigurationChanges(); From 34dc97ed95480bc7e8ffef5df99941fa2d07c3ed Mon Sep 17 00:00:00 2001 From: SummerC0zyR0ck Date: Fri, 18 Sep 2026 01:48:17 +0000 Subject: [PATCH 4/5] test(runtime-host): assert the Artifact epoch moved past 177 --- packages/runtime-host/src/__tests__/protocol.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index d00eaf65ce..52570183d0 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -135,7 +135,7 @@ describe('Runtime Host bootstrap protocol', () => { }); test('publishes a new compatibility epoch for Artifact invalidation frames', () => { - assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 161); + assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 177); }); test('publishes a new compatibility epoch for mandatory submit Skill outcomes', () => { From ac37976123abfdc7c240f31c498323d933bc0d77 Mon Sep 17 00:00:00 2001 From: SummerC0zyR0ck Date: Thu, 24 Sep 2026 07:32:06 +0000 Subject: [PATCH 5/5] fix(runtime-host): advance artifact invalidation epoch --- packages/runtime-host/src/__tests__/protocol.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index 52570183d0..064ca51937 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -135,7 +135,7 @@ describe('Runtime Host bootstrap protocol', () => { }); test('publishes a new compatibility epoch for Artifact invalidation frames', () => { - assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 177); + assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 183); }); test('publishes a new compatibility epoch for mandatory submit Skill outcomes', () => {