diff --git a/src/main/agent-hooks/server/server-listeners.ts b/src/main/agent-hooks/server/server-listeners.ts index 60781aef03f..1a1fe52e41f 100644 --- a/src/main/agent-hooks/server/server-listeners.ts +++ b/src/main/agent-hooks/server/server-listeners.ts @@ -21,6 +21,9 @@ import { AgentHookServerState } from './server-state' import { serializeAgentStatusSubject } from '../../../shared/agent-status-subject' import { structuredStatusLegacyEvent } from './server-structured-status-row' +// Why: the listing counter starts at 1, so an unassigned row must sort last — never above every ordered row. +const UNORDERED_STATUS_ROW = Number.MAX_SAFE_INTEGER + export abstract class AgentHookServerListeners extends AgentHookServerState { protected emitEnrichedStatus(enriched: EnrichedAgentHookEventPayload): void { this.onAgentStatus?.(enriched) @@ -47,7 +50,7 @@ export abstract class AgentHookServerListeners extends AgentHookServerState { rows.push({ // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Main admits enriched legacy rows; shared listeners expose only the base event type. entry: entry as EnrichedAgentHookEventPayload, - order: getLegacyStatusListingOrder(this.state, paneKey) ?? 0 + order: getLegacyStatusListingOrder(this.state, paneKey) ?? UNORDERED_STATUS_ROW }) } for (const parent of this.canonicalStatusStore.getSnapshot().parents) { @@ -56,7 +59,9 @@ export abstract class AgentHookServerListeners extends AgentHookServerState { } rows.push({ entry: structuredStatusLegacyEvent(parent.status), - order: this.canonicalListingOrder.get(serializeAgentStatusSubject(parent.subject)) ?? 0 + order: + this.canonicalListingOrder.get(serializeAgentStatusSubject(parent.subject)) ?? + UNORDERED_STATUS_ROW }) } return rows.sort((a, b) => a.order - b.order).map(({ entry }) => entry) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-ownership.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-ownership.test.ts index b795b899567..7e9e6da84a3 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-status-ownership.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-ownership.test.ts @@ -53,6 +53,23 @@ describe('structured status owner address retention', () => { ) }) + it('does not report a throwing publication as an owned location', () => { + const sink = { + publish: vi.fn().mockImplementationOnce(() => { + throw new Error('store down') + }), + forget: vi.fn() + } + const owner = new StructuredAgentSessionStatusOwnership(() => sink) + expect(() => owner.publish(summary, location)).toThrow('store down') + // The feed skips an unchanged re-projection only when the location already matches. Reporting a + // match here would strand the row: the publish never landed and nothing else re-offers it. + expect(owner.matchesLocation(summary.sessionId, location)).toBe(false) + owner.publish(summary, location) + expect(sink.publish).toHaveBeenCalledTimes(2) + expect(owner.matchesLocation(summary.sessionId, location)).toBe(true) + }) + it('keeps the owner address when a downstream publication observer throws', () => { const sink = { publish: vi.fn(() => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-ownership.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-ownership.ts index 8354d23e523..088879f8ddc 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-status-ownership.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-ownership.ts @@ -17,12 +17,17 @@ export type StructuredAgentSessionStatusSink = { /** Retain the owner address because record removal may precede the final status callback. */ export class StructuredAgentSessionStatusOwnership { private readonly subjects = new Map() + // Why separate from `subjects`: the address must survive a throwing publish so teardown can still + // forget a row that did land, but "we hold an address" is not evidence the row is there. Only a + // publish that returned proves that, and only that proof may suppress the re-offer below. + private readonly landed = new Set() constructor(private readonly sink: () => StructuredAgentSessionStatusSink | undefined) {} matchesLocation(sessionId: string, location: AgentSessionExecutionLocation): boolean { const subject = this.subjects.get(sessionId) return ( + this.landed.has(sessionId) && subject?.executionHostId === location.executionHostId && subject.wslDistro === location.wslDistro && subject.workspaceId === location.workspaceId && @@ -53,7 +58,9 @@ export class StructuredAgentSessionStatusOwnership { sink.forget(previous) } this.subjects.set(summary.sessionId, subject) + this.landed.delete(summary.sessionId) sink.publish(summary, subject) + this.landed.add(summary.sessionId) } forget(sessionId: string): void { @@ -61,6 +68,7 @@ export class StructuredAgentSessionStatusOwnership { if (!subject) { return } + this.landed.delete(sessionId) this.sink()?.forget(subject) this.subjects.delete(sessionId) } diff --git a/src/shared/agent-status-store-reopen.test.ts b/src/shared/agent-status-store-reopen.test.ts index 633d6e9a5b6..daab09e8bf1 100644 --- a/src/shared/agent-status-store-reopen.test.ts +++ b/src/shared/agent-status-store-reopen.test.ts @@ -1,6 +1,5 @@ import { describe, expect, it } from 'vitest' import { createAgentStatusStore } from './agent-status-store' -import { AGENT_STATUS_STORE_LIMITS } from './agent-status-store-contract' import { makePtyRunAgentStatusSubject, makeStructuredAgentStatusSubject @@ -15,7 +14,7 @@ const scope = { const subject = makeStructuredAgentStatusSubject(scope, 'durable-session') describe('structured parent reopening', () => { - it('reopens a removed structured parent and fences replay from before the reopen', () => { + it('reopens a removed structured parent and refuses a replay whose revision pair is spent', () => { const owner = createAgentStatusStore({ epoch: 'host', mode: 'authority' }) const replica = createAgentStatusStore({ epoch: 'reader', mode: 'replica' }) const oldPublication = owner.applyMutation({ parent: { subject, firstObservedAt: 10 } }) @@ -24,27 +23,25 @@ describe('structured parent reopening', () => { const removal = owner.applyMutation({ removeParent: subject }) expect(removal).not.toBeNull() expect(replica.applyTransportEnvelope(removal)).toBe(true) + expect(replica.getParent(subject)).toBeNull() const reopened = owner.applyMutation({ parent: { subject, firstObservedAt: 30 } }) expect(reopened).not.toBeNull() expect(replica.applyTransportEnvelope(reopened)).toBe(true) expect(replica.getParent(subject)).toEqual(owner.getParent(subject)) - expect(replica.applyTransportEnvelope(oldPublication)).toBe(false) - expect(replica.getSnapshot()).toEqual(owner.getSnapshot()) + expect(replica.getParent(subject)?.firstObservedAt).toBe(30) - // The revision envelope, not the tombstone's presence, is what fences the stale replay. - expect( - owner.applyMutation({ - removeChildren: Array.from( - { length: AGENT_STATUS_STORE_LIMITS.tombstones + 1 }, - (_, index) => `unused-child-${index}` - ) - }) - ).not.toBeNull() - expect(owner.getSnapshot().tombstones.some((item) => item.entity === 'parent')).toBe(false) - expect(replica.applySnapshot(owner.getSnapshot())).toBe(true) + // Outcome only, deliberately: a spent replay is refused and a resequenced one is not. Which + // layer refuses it is NOT asserted, because no test at this API can tell — transport + // consecutiveness, the parent-revision validator and the tombstone guard each refuse it alone, + // and ablating any two leaves this green. Attributing one of them here would be a false claim. expect(replica.applyTransportEnvelope(oldPublication)).toBe(false) expect(replica.getParent(subject)?.firstObservedAt).toBe(30) + const resequenced = owner.applyMutation({ parent: { subject, firstObservedAt: 10 } }) + expect(resequenced).not.toBeNull() + expect(replica.applyTransportEnvelope(resequenced)).toBe(true) + expect(replica.getParent(subject)?.firstObservedAt).toBe(10) + expect(replica.getSnapshot()).toEqual(owner.getSnapshot()) }) it('fences a republication only inside the removing mutation, for every subject kind', () => {