From 20db8b76cc7d59e3ff361eb245e9471fe7ab5dc8 Mon Sep 17 00:00:00 2001 From: SummerC0zyR0ck Date: Wed, 16 Sep 2026 09:13:13 +0000 Subject: [PATCH 1/8] 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 | 25 +++++++ .../server/session-retirement-coordinator.ts | 4 + .../server/session-revision-coordinator.ts | 4 + .../src/server/session-sidecar-purge.ts | 8 +- 28 files changed, 661 insertions(+), 10 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 2d7b3b0911..2f52baf456 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 aaa0d4ca4a..2fbf289da2 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -1562,6 +1562,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"); }); @@ -1841,6 +1848,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 c3e7507961..5b4a2f7d7a 100644 --- a/apps/desktop/src/main/runtime-host-client.ts +++ b/apps/desktop/src/main/runtime-host-client.ts @@ -102,6 +102,7 @@ import { type ProjectDirectoryEntry, type ProjectDirectoryRoot, SESSION_TRANSCRIPT_BOOTSTRAP_MAX_BYTES, + type ArtifactChangedFrame, type SessionCatalogChangedFrame, type ScheduledTaskChangedFrame, type SessionCatalogItem, @@ -393,6 +394,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 c6896a2b96..cfc5df2a72 100644 --- a/packages/cli/src/__tests__/acp-stdio-server.test.ts +++ b/packages/cli/src/__tests__/acp-stdio-server.test.ts @@ -708,6 +708,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 2755642f16..fd64297b74 100644 --- a/packages/cli/src/__tests__/runtime-host-onboarding.test.ts +++ b/packages/cli/src/__tests__/runtime-host-onboarding.test.ts @@ -164,6 +164,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 fb350b705c..07e3ab4870 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(); @@ -70,6 +85,27 @@ test('routes each change kind only to subscribed connections', () => { feed.publishSessionCatalog('session-1', { kind: 'completed', eventId: 'terminal-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'], @@ -79,7 +115,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(all.slice(3, 5), [ { kind: 'session.catalog.changed', revision: 1, sessionId: 'session-1', attention }, { diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index 98692b175f..5f79d0b1b3 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 > 196); + }); + test('publishes a new compatibility epoch for durable external turn origins', () => { assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 189); }); 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 b363c9823d..7ff4efbe87 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>(); @@ -733,6 +736,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); @@ -766,6 +774,9 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { continue; } switch (frame.kind) { + case 'artifact.changed': + this.#acceptArtifactChanged(frame); + continue; case 'configuration.changed': this.#acceptConfigurationChanged(frame); continue; @@ -848,6 +859,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 { @@ -1053,6 +1074,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 2698e6a648..c63f870123 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'; @@ -104,7 +106,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 = 200 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 201 as const; +// 201: Host change feed includes Artifact deletion and Session purge invalidation frames. // 200: Agent Graph operator snapshots carry bounded output previews and metrics. // Older peers reject the additional `output` field on strict operator shapes. // 199: `storage.usage.query` and `storage.usage.sessions.query` report storage @@ -564,6 +567,7 @@ export type HostFrame = | SubscriptionFrame | ClientCapabilityHostFrame | ConfigurationChangedFrame + | ArtifactChangedFrame | ConnectionCatalogChangedFrame | ProjectCatalogChangedFrame | SessionCatalogChangedFrame @@ -698,6 +702,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 4022795336..2f602ae19e 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -1736,6 +1736,7 @@ export async function createExecutionRuntimeHostComposition( stores.sessionStore, Date.now, context.sessionAccessAuthority, + (sessionId, artifactId) => hostChanges.publishArtifactDeleted(sessionId, artifactId), ); rootCoordinator = new RootTurnCoordinator( manager, @@ -2751,6 +2752,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, @@ -2784,6 +2786,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 93a40ae850..ea3756a6ef 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, @@ -28,6 +29,7 @@ import type { } from '../protocol/index.js'; export type HostChangeFrame = + | ArtifactChangedFrame | ConfigurationChangedFrame | ConnectionCatalogChangedFrame | ProjectCatalogChangedFrame @@ -39,6 +41,7 @@ export interface HostChangeSubscription { } export interface HostChangeSubscriptionMask { + readonly artifact?: true | { readonly sessionId: string }; readonly configuration?: boolean; readonly connectionCatalog?: boolean; readonly projectCatalog?: boolean; @@ -86,6 +89,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; @@ -153,6 +164,15 @@ export class HostChangeFeed { subscription.mask.sessionCatalog?.sessionId === frame.sessionId && subscription.mask.sessionCatalog.principalId === closeScopeFor && this.#subscriptions.get(connectionId) === subscription + ) { + this.#subscriptions.delete(connectionId); + continue; + } + if ( + closeScopeFor !== undefined && + subscription.mask.artifact !== true && + subscription.mask.artifact?.sessionId === frame.sessionId && + this.#subscriptions.get(connectionId) === subscription ) { this.#subscriptions.delete(connectionId); } @@ -162,6 +182,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 09f1aab0b0..eae963c48e 100644 --- a/packages/runtime-host/src/server/session-revision-coordinator.ts +++ b/packages/runtime-host/src/server/session-revision-coordinator.ts @@ -113,6 +113,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. */ @@ -770,6 +771,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 36648543e7542e2fdc506fb499a6320569b25dd5 Mon Sep 17 00:00:00 2001 From: SummerC0zyR0ck Date: Thu, 17 Sep 2026 02:59:33 +0000 Subject: [PATCH 2/8] 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 0be40cd9e4..87e89a365c 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 @@ -62,6 +62,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'; @@ -522,6 +523,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 4615a6b4c64377e41bf019a205cf93f60e9fee94 Mon Sep 17 00:00:00 2001 From: SummerC0zyR0ck Date: Thu, 17 Sep 2026 03:33:18 +0000 Subject: [PATCH 3/8] 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 87e89a365c..9052df6590 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 @@ -523,47 +523,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 2fbf289da2..cf09bd2fcb 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -1842,6 +1842,7 @@ function registerHostClientIpc( }); registerOnboardingIpc({ onboardingService, ipcMain: scopedIpc }); registerTaskSubmissionReadinessIpc(taskSubmissionReadinessService, scopedIpc); + managedArtifactPreview.openScope(scope.targetEpoch); return async () => { clientPluginTransport.release(client); unsubscribeConfigurationChanges(); From 7e0a17caab6bf2082ca016f6f418c4b2de8f088c Mon Sep 17 00:00:00 2001 From: SummerC0zyR0ck Date: Mon, 28 Sep 2026 01:15:37 +0000 Subject: [PATCH 4/8] fix(runtime-host): resolve artifact preview rebase conflicts --- .../src/__tests__/host-change-feed.test.ts | 71 +++++++++---------- .../src/server/host-change-feed.ts | 1 + 2 files changed, 35 insertions(+), 37 deletions(-) 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 07e3ab4870..98b59df7a1 100644 --- a/packages/runtime-host/src/__tests__/host-change-feed.test.ts +++ b/packages/runtime-host/src/__tests__/host-change-feed.test.ts @@ -23,23 +23,11 @@ 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 }, @@ -53,7 +41,6 @@ test('routes each change kind only to subscribed connections', () => { feed.attachConnection( 'all', { - artifact: true, configuration: true, connectionCatalog: true, projectCatalog: true, @@ -73,8 +60,6 @@ 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(); @@ -85,27 +70,6 @@ test('routes each change kind only to subscribed connections', () => { feed.publishSessionCatalog('session-1', { kind: 'completed', eventId: 'terminal-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'], @@ -115,7 +79,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, 10); + assert.equal(all.length, 8); assert.deepEqual(all.slice(3, 5), [ { kind: 'session.catalog.changed', revision: 1, sessionId: 'session-1', attention }, { @@ -132,6 +96,39 @@ test('routes each change kind only to subscribed connections', () => { assert.equal(otherGuest.length, 3); }); +test('routes Artifact invalidations by Session scope', () => { + const feed = new HostChangeFeed(); + const all: unknown[] = []; + const scoped: unknown[] = []; + feed.attachConnection( + 'all-artifacts', + { artifact: true }, + { + send: async (frame) => void all.push(frame), + }, + ); + feed.attachConnection( + 'session-artifacts', + { artifact: { sessionId: 'session-1' } }, + { + send: async (frame) => void scoped.push(frame), + }, + ); + + feed.publishArtifactDeleted('session-1', 'artifact-1'); + feed.publishArtifactSessionPurged('session-2'); + + assert.equal(all.length, 2); + assert.deepEqual(scoped, [ + { + kind: 'artifact.changed', + reason: 'deleted', + sessionId: 'session-1', + artifactId: 'artifact-1', + }, + ]); +}); + test('keeps catalog revisions independent and removes failed subscriptions', async () => { const feed = new HostChangeFeed(); const frames: unknown[] = []; diff --git a/packages/runtime-host/src/server/host-change-feed.ts b/packages/runtime-host/src/server/host-change-feed.ts index ea3756a6ef..183404f61c 100644 --- a/packages/runtime-host/src/server/host-change-feed.ts +++ b/packages/runtime-host/src/server/host-change-feed.ts @@ -170,6 +170,7 @@ export class HostChangeFeed { } if ( closeScopeFor !== undefined && + frame.kind === 'session.catalog.changed' && subscription.mask.artifact !== true && subscription.mask.artifact?.sessionId === frame.sessionId && this.#subscriptions.get(connectionId) === subscription From 3128ab650e468f3aa24bd79050aea80b90514fa9 Mon Sep 17 00:00:00 2001 From: SummerC0zyR0ck Date: Mon, 28 Sep 2026 06:50:45 +0000 Subject: [PATCH 5/8] fix(runtime-host): isolate artifact invalidations from guests --- .../main/__tests__/main-startup-lifetime.test.ts | 11 +++++++++++ .../__tests__/managed-artifact-preview.test.ts | 4 ++-- apps/desktop/src/main/managed-artifact-preview.ts | 8 -------- .../src/__tests__/connection-session.test.ts | 8 ++++---- .../src/__tests__/host-change-feed.test.ts | 5 ++++- .../session-retirement-coordinator.test.ts | 10 +++++++--- packages/runtime-host/src/protocol/index.ts | 1 + .../runtime-host/src/server/connection-session.ts | 14 +++++++------- .../runtime-host/src/server/host-change-feed.ts | 9 --------- .../src/server/session-sidecar-purge.ts | 6 +++--- 10 files changed, 39 insertions(+), 37 deletions(-) diff --git a/apps/desktop/src/main/__tests__/main-startup-lifetime.test.ts b/apps/desktop/src/main/__tests__/main-startup-lifetime.test.ts index db1a6082dc..f8fb8e3fa0 100644 --- a/apps/desktop/src/main/__tests__/main-startup-lifetime.test.ts +++ b/apps/desktop/src/main/__tests__/main-startup-lifetime.test.ts @@ -124,6 +124,17 @@ test('does not release renderer IPC before persistent handlers are registered', assert.doesNotMatch(readyBody, /runtimeHostManager\?\.start\(\)/u); }); +test('wires Artifact deletion and purge events to preview invalidation for the active scope', () => { + const subscribe = bootSource.indexOf('client.subscribeArtifactChanges('); + const unsubscribe = bootSource.indexOf('unsubscribeArtifactChanges();', subscribe); + assert.ok(subscribe >= 0 && unsubscribe > subscribe); + const handler = bootSource.slice(subscribe, unsubscribe); + assert.match(handler, /managedArtifactPreview\.revoke\(scope\.targetEpoch,\s*frame\.sessionId,\s*frame\.artifactId\)/u); + assert.match(handler, /managedArtifactPreview\.releaseSession\(scope\.targetEpoch,\s*frame\.sessionId\)/u); + assert.match(bootSource, /managedArtifactPreview\.openScope\(scope\.targetEpoch\)/u); + assert.match(bootSource, /await managedArtifactPreview\.closeScope\(scope\.targetEpoch\)/u); +}); + test('resolves persisted locale before first post-settings recovery prompt', () => { const rendererRecoveryStart = earlyWindowSource.indexOf('onRendererProcessGone: async'); const rendererRecovery = earlyWindowSource.slice( 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 65a3fd12ad..1807e9c81a 100644 --- a/apps/desktop/src/main/__tests__/managed-artifact-preview.test.ts +++ b/apps/desktop/src/main/__tests__/managed-artifact-preview.test.ts @@ -158,7 +158,7 @@ test('bounds previews per session instead of starving another session', async () } }); -test('evicts the oldest lease at the global backstop without denying another session', async () => { +test('does not invalidate existing previews when another session prepares one', async () => { const service = new ManagedArtifactPreview(); try { const endpoints = []; @@ -166,7 +166,7 @@ test('evicts the oldest lease at the global backstop without denying another ses 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(endpoints[0]!.url)).text(), 'preview-0'); assert.equal(await (await fetch(replacement.url)).text(), 'replacement'); } finally { await service.close(); diff --git a/apps/desktop/src/main/managed-artifact-preview.ts b/apps/desktop/src/main/managed-artifact-preview.ts index c26d00ce87..48933cff27 100644 --- a/apps/desktop/src/main/managed-artifact-preview.ts +++ b/apps/desktop/src/main/managed-artifact-preview.ts @@ -24,10 +24,6 @@ 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; @@ -82,10 +78,6 @@ export class ManagedArtifactPreview { 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() }; diff --git a/packages/runtime-host/src/__tests__/connection-session.test.ts b/packages/runtime-host/src/__tests__/connection-session.test.ts index 5b3b029ef1..0d888d9a20 100644 --- a/packages/runtime-host/src/__tests__/connection-session.test.ts +++ b/packages/runtime-host/src/__tests__/connection-session.test.ts @@ -137,7 +137,7 @@ test('concurrent responses remain framed and correlated in reverse completion or ); }); -test('scopes Session Guest Artifact changes to the shared Session', async () => { +test('does not expose Artifact invalidations to Session Guests', async () => { const pair = await openTransportPair(); const hostChanges = new HostChangeFeed(); const session = new RuntimeHostConnectionSession({ @@ -184,11 +184,11 @@ test('scopes Session Guest Artifact changes to the shared Session', async () => try { hostChanges.publishArtifactDeleted('session-2', 'artifact-2'); hostChanges.publishArtifactDeleted('session-1', 'artifact-1'); + hostChanges.publishSessionCatalog('session-1'); assert.deepEqual(decodeHostFrame(await pair.clientTransport.read(1_000)), { - kind: 'artifact.changed', - reason: 'deleted', + kind: 'session.catalog.changed', + revision: 1, sessionId: 'session-1', - artifactId: 'artifact-1', }); } finally { pair.clientTransport.abort(); 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 98b59df7a1..e0ffda3a10 100644 --- a/packages/runtime-host/src/__tests__/host-change-feed.test.ts +++ b/packages/runtime-host/src/__tests__/host-change-feed.test.ts @@ -56,7 +56,10 @@ test('routes each change kind only to subscribed connections', () => { ); feed.attachConnection( 'other-guest', - { sessionCatalog: { sessionId: 'session-1', principalId: 'guest-2' } }, + { + artifact: { sessionId: 'session-1' }, + sessionCatalog: { sessionId: 'session-1', principalId: 'guest-2' }, + }, { send: async (frame) => void otherGuest.push(frame) }, ); 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 fde5f7e3a8..2a0c0f64e6 100644 --- a/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts @@ -196,7 +196,7 @@ describe('Host Session retirement coordinator', () => { assert.equal(garbageBatches, 0); }); - test('publishes Session Artifact invalidation only when Artifact purge succeeds', async () => { + test('publishes Session Artifact invalidation even when a purge partially fails', async () => { const purged: string[] = []; await purgeSessionSidecars( { @@ -225,7 +225,7 @@ describe('Host Session retirement coordinator', () => { ), AggregateError, ); - assert.deepEqual(purged, ['session-success']); + assert.deepEqual(purged, ['session-success', 'session-artifact-failure']); await assert.rejects( purgeSessionSidecars( @@ -243,7 +243,11 @@ describe('Host Session retirement coordinator', () => { ), AggregateError, ); - assert.deepEqual(purged, ['session-success', 'session-other-failure']); + assert.deepEqual(purged, [ + 'session-success', + 'session-artifact-failure', + 'session-other-failure', + ]); }); test('rejects ordinary archive and remove operations for the Coordination Session', async () => { diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index c63f870123..2fc16af816 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -110,6 +110,7 @@ export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 201 as const; // 201: Host change feed includes Artifact deletion and Session purge invalidation frames. // 200: Agent Graph operator snapshots carry bounded output previews and metrics. // Older peers reject the additional `output` field on strict operator shapes. +// 201: Artifact invalidation feed and guest routing require matching peers. // 199: `storage.usage.query` and `storage.usage.sessions.query` report storage // usage. An epoch-198 Host rejects the unknown operation and drops the // connection, so the pair must fail admission instead. diff --git a/packages/runtime-host/src/server/connection-session.ts b/packages/runtime-host/src/server/connection-session.ts index 48537b05b3..001a7b89aa 100644 --- a/packages/runtime-host/src/server/connection-session.ts +++ b/packages/runtime-host/src/server/connection-session.ts @@ -366,14 +366,14 @@ export class RuntimeHostConnectionSession { : hasRuntimeHostOperationGrant(this.#options.connection.authority, 'session.catalog.query') ? true : undefined; + // Artifact invalidations are consumed by the owner Desktop. Guest clients + // must not learn Artifact identities or share a session-scoped subscription + // whose lifetime is tied to another guest's catalog scope. 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.#options.connection.authority.principalKind !== 'session_guest' && + hasRuntimeHostOperationGrant(this.#options.connection.authority, 'artifact.query') + ? true + : undefined; this.#hostChanges = service.attachConnection( this.#options.connection.connectionId, { diff --git a/packages/runtime-host/src/server/host-change-feed.ts b/packages/runtime-host/src/server/host-change-feed.ts index 183404f61c..4ebbb33d71 100644 --- a/packages/runtime-host/src/server/host-change-feed.ts +++ b/packages/runtime-host/src/server/host-change-feed.ts @@ -168,15 +168,6 @@ export class HostChangeFeed { this.#subscriptions.delete(connectionId); continue; } - if ( - closeScopeFor !== undefined && - frame.kind === 'session.catalog.changed' && - subscription.mask.artifact !== true && - subscription.mask.artifact?.sessionId === frame.sessionId && - this.#subscriptions.get(connectionId) === subscription - ) { - this.#subscriptions.delete(connectionId); - } } } } diff --git a/packages/runtime-host/src/server/session-sidecar-purge.ts b/packages/runtime-host/src/server/session-sidecar-purge.ts index 35d900eff6..7e9f6ac899 100644 --- a/packages/runtime-host/src/server/session-sidecar-purge.ts +++ b/packages/runtime-host/src/server/session-sidecar-purge.ts @@ -39,9 +39,9 @@ export async function purgeSessionSidecars( ...(authority.contextOffload ? [authority.contextOffload.retireSession(sessionId)] : []), authority.purgeOperationalState(sessionId), ]); - // 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); + // Invalidation is idempotent and must happen even when the Artifact purge + // rejects: a failed purge may have removed only part of the Session's data. + authority.onArtifactsPurged?.(sessionId); const failures = [artifactPurge, ...sidecarPurges].flatMap((outcome) => outcome.status === 'rejected' ? [outcome.reason] : [], ); From 14b96053f66c67bf4a179d58a34364e74de83158 Mon Sep 17 00:00:00 2001 From: SummerC0zyR0ck Date: Mon, 28 Sep 2026 07:57:13 +0000 Subject: [PATCH 6/8] fix(desktop): bound aggregate artifact previews --- .../managed-artifact-preview.test.ts | 29 +++++++++++++++++-- .../src/main/managed-artifact-preview.ts | 19 ++++++++++-- 2 files changed, 43 insertions(+), 5 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 1807e9c81a..08e6804ae9 100644 --- a/apps/desktop/src/main/__tests__/managed-artifact-preview.test.ts +++ b/apps/desktop/src/main/__tests__/managed-artifact-preview.test.ts @@ -158,21 +158,44 @@ test('bounds previews per session instead of starving another session', async () } }); -test('does not invalidate existing previews when another session prepares one', async () => { +test('rejects previews above the global limit without invalidating existing sessions', 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(service.prepare('h', client('replacement'), 's64', 'a1'), /across the Desktop/); assert.equal(await (await fetch(endpoints[0]!.url)).text(), 'preview-0'); - assert.equal(await (await fetch(replacement.url)).text(), 'replacement'); + assert.equal(await (await fetch(endpoints[63]!.url)).text(), 'preview-63'); } finally { await service.close(); } }); +test('reserves aggregate preview bytes before streaming and releases reservations on failure', async () => { + const service = new ManagedArtifactPreview(); + let resume!: () => void; + const gate = new Promise((resolve) => { resume = resolve; }); + const large = { + getArtifact: async () => ({ id: 'a1', sessionId: 's1', turnId: 't1', createdAt: 0, name: 'large.html', kind: 'html' as const, sizeBytes: PREVIEW_MAX_BYTES, source: 'tool_result' as const }), + streamArtifact: async () => { await gate; throw new Error('test stream failure'); }, + }; + const pending = Array.from({ length: 16 }, (_, index) => service.prepare('h', large, `s${index + 1}`, 'a1')); + try { + // Each in-flight Artifact reserves 8 MiB, reaching the 128 MiB aggregate budget. + await new Promise((resolve) => setImmediate(resolve)); + await assert.rejects(service.prepare('h', large, 's17', 'a1'), /memory limit/); + resume(); + await Promise.all(pending.map((preparation) => assert.rejects(preparation, /test stream failure/))); + assert.equal((await service.prepare('h', client(), 's1', 'a1')).reachable, true); + } finally { + resume(); + await Promise.allSettled(pending); + await service.close(); + } +}); + test('releases every preview for a purged session', async () => { const service = new ManagedArtifactPreview(); try { diff --git a/apps/desktop/src/main/managed-artifact-preview.ts b/apps/desktop/src/main/managed-artifact-preview.ts index 48933cff27..f33fc2cd65 100644 --- a/apps/desktop/src/main/managed-artifact-preview.ts +++ b/apps/desktop/src/main/managed-artifact-preview.ts @@ -24,6 +24,8 @@ import type { DesktopRuntimeHostClient } from './runtime-host-client.js'; export const PREVIEW_MAX_BYTES = 8 * 1024 * 1024; const MAX_PREVIEWS = 16; +const MAX_GLOBAL_PREVIEWS = 64; +const MAX_GLOBAL_PREVIEW_BYTES = 128 * 1024 * 1024; const PREVIEW_TTL_MS = 30 * 60 * 1000; const READ_DEADLINE_MS = 30_000; @@ -42,12 +44,14 @@ interface Lease { server: Server; timer?: ReturnType; url?: string; + reservedBytes: number; } /** Desktop-owned, bounded, ephemeral HTML snapshots. No workspace directory is served. */ export class ManagedArtifactPreview { private readonly leases = new Set(); private readonly retiredScopes = new Set(); + private reservedBytes = 0; private closed = false; constructor(private readonly ttlMs = PREVIEW_TTL_MS) {} @@ -78,9 +82,12 @@ export class ManagedArtifactPreview { if (sessionLeases.length >= MAX_PREVIEWS) { throw new Error('Too many active previews; wait for expiry'); } + if (this.leases.size >= MAX_GLOBAL_PREVIEWS) { + throw new Error('Too many active previews across the Desktop; wait for expiry'); + } signal?.throwIfAborted(); // Reserve before asynchronous reads, so concurrent preparations cannot exceed the bound. - const lease: Lease = { scope, sessionId, artifactId, server: createServer() }; + const lease: Lease = { scope, sessionId, artifactId, server: createServer(), reservedBytes: 0 }; this.leases.add(lease); const assertActive = () => { signal?.throwIfAborted(); @@ -93,6 +100,11 @@ export class ManagedArtifactPreview { if (!Number.isSafeInteger(artifact.sizeBytes) || artifact.sizeBytes < 0 || artifact.sizeBytes > PREVIEW_MAX_BYTES) { throw new Error('HTML preview exceeds the 8 MiB limit; use Save As instead'); } + if (this.reservedBytes + artifact.sizeBytes > MAX_GLOBAL_PREVIEW_BYTES) { + throw new Error('The Desktop preview memory limit has been reached; wait for previews to expire'); + } + lease.reservedBytes = artifact.sizeBytes; + this.reservedBytes += lease.reservedBytes; const chunks: Buffer[] = []; let size = 0; const total = await withDeadline(client.streamArtifact(sessionId, artifactId, async (chunk) => { @@ -184,7 +196,10 @@ export class ManagedArtifactPreview { } private async release(lease: Lease): Promise { - this.leases.delete(lease); + if (this.leases.delete(lease)) { + this.reservedBytes -= lease.reservedBytes; + lease.reservedBytes = 0; + } clearTimeout(lease.timer); await new Promise((resolve) => { lease.server.close(() => resolve()); From 061be7c79e5656a66005fb8e081a83b860edb962 Mon Sep 17 00:00:00 2001 From: SummerC0zyR0ck Date: Mon, 28 Sep 2026 09:34:07 +0000 Subject: [PATCH 7/8] fix(runtime-host): align artifact preview review fixes --- .../__tests__/main-startup-lifetime.test.ts | 4 ++++ .../managed-artifact-preview.test.ts | 15 ++++++++++++- .../src/__tests__/host-change-feed.test.ts | 22 ++++--------------- .../src/server/host-change-feed.ts | 8 ++----- 4 files changed, 24 insertions(+), 25 deletions(-) diff --git a/apps/desktop/src/main/__tests__/main-startup-lifetime.test.ts b/apps/desktop/src/main/__tests__/main-startup-lifetime.test.ts index f8fb8e3fa0..ef765f2685 100644 --- a/apps/desktop/src/main/__tests__/main-startup-lifetime.test.ts +++ b/apps/desktop/src/main/__tests__/main-startup-lifetime.test.ts @@ -133,6 +133,10 @@ test('wires Artifact deletion and purge events to preview invalidation for the a assert.match(handler, /managedArtifactPreview\.releaseSession\(scope\.targetEpoch,\s*frame\.sessionId\)/u); assert.match(bootSource, /managedArtifactPreview\.openScope\(scope\.targetEpoch\)/u); assert.match(bootSource, /await managedArtifactPreview\.closeScope\(scope\.targetEpoch\)/u); + assert.ok( + bootSource.indexOf('await managedArtifactPreview.closeScope(scope.targetEpoch)') < + bootSource.indexOf('unsubscribeArtifactChanges();'), + ); }); test('resolves persisted locale before first post-settings recovery prompt', () => { 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 08e6804ae9..344a14a01d 100644 --- a/apps/desktop/src/main/__tests__/managed-artifact-preview.test.ts +++ b/apps/desktop/src/main/__tests__/managed-artifact-preview.test.ts @@ -168,6 +168,9 @@ test('rejects previews above the global limit without invalidating existing sess await assert.rejects(service.prepare('h', client('replacement'), 's64', 'a1'), /across the Desktop/); assert.equal(await (await fetch(endpoints[0]!.url)).text(), 'preview-0'); assert.equal(await (await fetch(endpoints[63]!.url)).text(), 'preview-63'); + await service.releaseUrl(endpoints[0]!.url); + const replacement = await service.prepare('h', client('replacement'), 's64', 'a1'); + assert.equal(await (await fetch(replacement.url)).text(), 'replacement'); } finally { await service.close(); } @@ -181,6 +184,14 @@ test('reserves aggregate preview bytes before streaming and releases reservation getArtifact: async () => ({ id: 'a1', sessionId: 's1', turnId: 't1', createdAt: 0, name: 'large.html', kind: 'html' as const, sizeBytes: PREVIEW_MAX_BYTES, source: 'tool_result' as const }), streamArtifact: async () => { await gate; throw new Error('test stream failure'); }, }; + const successfulLarge = { + getArtifact: large.getArtifact, + streamArtifact: async (_sessionId: string, _artifactId: string, write: (chunk: Uint8Array) => Promise) => { + const bytes = Buffer.alloc(PREVIEW_MAX_BYTES); + await write(bytes); + return bytes.length; + }, + }; const pending = Array.from({ length: 16 }, (_, index) => service.prepare('h', large, `s${index + 1}`, 'a1')); try { // Each in-flight Artifact reserves 8 MiB, reaching the 128 MiB aggregate budget. @@ -188,7 +199,9 @@ test('reserves aggregate preview bytes before streaming and releases reservation await assert.rejects(service.prepare('h', large, 's17', 'a1'), /memory limit/); resume(); await Promise.all(pending.map((preparation) => assert.rejects(preparation, /test stream failure/))); - assert.equal((await service.prepare('h', client(), 's1', 'a1')).reachable, true); + const recovered = await service.prepare('h', successfulLarge, 's1', 'a1'); + assert.equal(recovered.reachable, true); + await service.releaseUrl(recovered.url); } finally { resume(); await Promise.allSettled(pending); 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 e0ffda3a10..2697713521 100644 --- a/packages/runtime-host/src/__tests__/host-change-feed.test.ts +++ b/packages/runtime-host/src/__tests__/host-change-feed.test.ts @@ -57,7 +57,6 @@ test('routes each change kind only to subscribed connections', () => { feed.attachConnection( 'other-guest', { - artifact: { sessionId: 'session-1' }, sessionCatalog: { sessionId: 'session-1', principalId: 'guest-2' }, }, { send: async (frame) => void otherGuest.push(frame) }, @@ -99,10 +98,10 @@ test('routes each change kind only to subscribed connections', () => { assert.equal(otherGuest.length, 3); }); -test('routes Artifact invalidations by Session scope', () => { +test('routes Artifact invalidations only to owner subscriptions', () => { const feed = new HostChangeFeed(); const all: unknown[] = []; - const scoped: unknown[] = []; + const guest: unknown[] = []; feed.attachConnection( 'all-artifacts', { artifact: true }, @@ -110,26 +109,13 @@ test('routes Artifact invalidations by Session scope', () => { send: async (frame) => void all.push(frame), }, ); - feed.attachConnection( - 'session-artifacts', - { artifact: { sessionId: 'session-1' } }, - { - send: async (frame) => void scoped.push(frame), - }, - ); + feed.attachConnection('guest', {}, { send: async (frame) => void guest.push(frame) }); feed.publishArtifactDeleted('session-1', 'artifact-1'); feed.publishArtifactSessionPurged('session-2'); assert.equal(all.length, 2); - assert.deepEqual(scoped, [ - { - kind: 'artifact.changed', - reason: 'deleted', - sessionId: 'session-1', - artifactId: 'artifact-1', - }, - ]); + assert.deepEqual(guest, []); }); test('keeps catalog revisions independent and removes failed subscriptions', async () => { diff --git a/packages/runtime-host/src/server/host-change-feed.ts b/packages/runtime-host/src/server/host-change-feed.ts index 4ebbb33d71..1b4cd38106 100644 --- a/packages/runtime-host/src/server/host-change-feed.ts +++ b/packages/runtime-host/src/server/host-change-feed.ts @@ -41,7 +41,7 @@ export interface HostChangeSubscription { } export interface HostChangeSubscriptionMask { - readonly artifact?: true | { readonly sessionId: string }; + readonly artifact?: true; readonly configuration?: boolean; readonly connectionCatalog?: boolean; readonly projectCatalog?: boolean; @@ -166,7 +166,6 @@ export class HostChangeFeed { this.#subscriptions.get(connectionId) === subscription ) { this.#subscriptions.delete(connectionId); - continue; } } } @@ -175,10 +174,7 @@ 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) - ); + return mask.artifact === true; case 'configuration.changed': return mask.configuration === true; case 'connection.catalog.changed': From 5813aea9a8548ba05934d9c7ed6acde4ec7a0873 Mon Sep 17 00:00:00 2001 From: SummerC0zyR0ck Date: Wed, 30 Sep 2026 08:44:59 +0000 Subject: [PATCH 8/8] fix(runtime-host): reconcile protocol epoch after rebase --- packages/runtime-host/src/protocol/index.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 2fc16af816..64089e7797 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -106,11 +106,11 @@ 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 = 201 as const; -// 201: Host change feed includes Artifact deletion and Session purge invalidation frames. +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 202 as const; +// 202: Host change feed includes Artifact deletion and Session purge invalidation frames; +// the Artifact invalidation feed and guest routing require matching peers. // 200: Agent Graph operator snapshots carry bounded output previews and metrics. // Older peers reject the additional `output` field on strict operator shapes. -// 201: Artifact invalidation feed and guest routing require matching peers. // 199: `storage.usage.query` and `storage.usage.sessions.query` report storage // usage. An epoch-198 Host rejects the unknown operation and drops the // connection, so the pair must fail admission instead.