diff --git a/.agentworkforce/agents/relay-feature-guardian/agent.test.ts b/.agentworkforce/agents/relay-feature-guardian/agent.test.ts index 792302f5c..af09878e6 100644 --- a/.agentworkforce/agents/relay-feature-guardian/agent.test.ts +++ b/.agentworkforce/agents/relay-feature-guardian/agent.test.ts @@ -1,37 +1,164 @@ import { readFileSync } from 'node:fs'; -import type { MemoryItem, WorkforceCtx } from '@agentworkforce/runtime'; +import type { WorkforceCtx } from '@agentworkforce/runtime'; import { bindPreviewTransport, + slackClient, type RelayTransport, type RelayTransportRequest, type RelayTransportWriteRequest, type WritebackResult, } from '@relayfile/relay-helpers'; import { describe, expect, it, vi } from 'vitest'; -import guardian, { deliveredSlackTs, featurePostIdempotencyKey, resolveManifestPath } from './agent.ts'; +import guardian, { + CYCLE_STATE_PATH, + ProgressStateConflictError, + SLACK_WRITEBACK_POLL_MS, + SLACK_WRITEBACK_TIMEOUT_MS, + createHttpProgressStore, + deliveredSlackTs, + featurePostIdempotencyKey, + resolveManifestPath, + runGuardian, + type ProgressState, +} from './agent.ts'; const persona = JSON.parse(readFileSync(new URL('./persona.json', import.meta.url), 'utf8')) as { inputs: { SLACK_CHANNEL: { default: string } }; }; -const manifest = ` -version: '1' -categories: - core: - name: Core - criticality: critical - features: - - id: start-broker - name: Start Broker - cli: relay node up - description: Starts the local broker. - verify_tier: 1 - - id: stop-broker - name: Stop Broker - cli: relay node down - description: Stops the local broker. - verify_tier: 1 -`; +const manifestFeatures = [ + { + id: 'broker-up', + name: 'Start Broker', + cli: 'relay node up', + description: 'Starts the local broker.', + tier: 1, + }, + { + id: 'broker-down', + name: 'Stop Broker', + cli: 'relay node down', + description: 'Stops the local broker.', + tier: 2, + }, + { + id: 'broker-status', + name: 'Broker Status', + cli: 'relay node status', + description: 'Shows broker status.', + tier: 1, + }, + ...Array.from({ length: 119 }, (_, index) => ({ + id: `feature-${index + 4}`, + name: `Feature ${index + 4}`, + cli: `relay feature-${index + 4}`, + description: `Checks feature ${index + 4}.`, + tier: 6, + })), +]; + +function renderManifest(features: typeof manifestFeatures): string { + return [ + "version: '1'", + 'categories:', + ' core:', + ' name: Core', + ' criticality: critical', + ' features:', + ...features.flatMap((feature) => [ + ` - id: ${feature.id}`, + ` name: ${feature.name}`, + ` cli: ${feature.cli}`, + ` description: ${feature.description}`, + ` verify_tier: ${feature.tier}`, + ]), + ].join('\n'); +} + +const manifest = renderManifest(manifestFeatures); + +const cycleStatePath = '/memory/workspace/relay-feature-guardian/cycle-state.json'; + +const storeFeatures = manifestFeatures.map((feature) => ({ + ...feature, + desc: feature.description, + criticality: 'critical' as const, +})); + +const orderedManifestFeatures = [...manifestFeatures].sort((a, b) => a.tier - b.tier); + +function progressState(checkedCount: number, generation = 1): ProgressState { + const checkedIds = orderedManifestFeatures.slice(0, checkedCount).map((feature) => feature.id); + return { + kind: 'relay-feature-guardian:progress', + version: 3, + generation, + checkedIds, + cycleStartedAt: generation === 1 ? '2026-07-18T10:26:47.981Z' : '2026-07-18T11:26:47.981Z', + totalFeatures: 122, + ...(checkedCount > 0 + ? { + lastPost: { + featureId: checkedIds.at(-1) as string, + ts: `1784370${checkedCount}.029509`, + }, + } + : {}), + }; +} + +class RelayfileStateServer { + content: string | null; + revision = 1; + failGetStatus = 0; + hangMethod: 'GET' | 'PUT' | null = null; + corruptReadBack = false; + readonly requests: Array<{ method: string; ifMatch: string | null }> = []; + + constructor(seed: ProgressState | null) { + this.content = seed ? `${JSON.stringify(seed)}\n` : null; + } + + readonly fetch = (async (input: URL | RequestInfo, init: RequestInit = {}) => { + const method = init.method ?? 'GET'; + const headers = new Headers(init.headers); + this.requests.push({ method, ifMatch: headers.get('if-match') }); + const url = new URL(input instanceof URL ? input : typeof input === 'string' ? input : input.url); + expect(url.searchParams.get('path')).toBe(CYCLE_STATE_PATH); + expect(headers.get('authorization')).toBe('Bearer relayfile-token'); + expect(headers.get('x-correlation-id')).toMatch(/^guardian-state-/); + + if (this.hangMethod === method) return new Promise(() => undefined); + if (method === 'GET') { + if (this.failGetStatus) { + return new Response(JSON.stringify({ error: 'read denied' }), { status: this.failGetStatus }); + } + if (this.content === null) return new Response('{}', { status: 404 }); + const content = this.corruptReadBack ? `${JSON.stringify(progressState(1))}\n` : this.content; + return new Response( + JSON.stringify({ + path: CYCLE_STATE_PATH, + revision: `rev-${this.revision}`, + content, + encoding: 'utf-8', + }), + { status: 200, headers: { ETag: `rev-${this.revision}` } } + ); + } + + expect(method).toBe('PUT'); + const ifMatch = headers.get('if-match'); + const expected = this.content === null ? '0' : `rev-${this.revision}`; + if (ifMatch !== expected) return new Response(JSON.stringify({ error: 'conflict' }), { status: 409 }); + this.content = String(init.body); + this.revision += 1; + return new Response(JSON.stringify({ revision: `rev-${this.revision}` }), { status: 200 }); + }) as typeof fetch; + + state(): ProgressState { + return JSON.parse(this.content ?? '{}') as ProgressState; + } +} class IdempotentSlackTransport implements RelayTransport { readonly attempts: RelayTransportWriteRequest[] = []; @@ -66,13 +193,41 @@ class IdempotentSlackTransport implements RelayTransport { } } -function guardianContext(failSaveCall: number): { +class DelayedSlackTransport extends IdempotentSlackTransport { + constructor(private readonly delayMs: number) { + super(); + } + + override async write(request: RelayTransportWriteRequest): Promise { + await new Promise((resolve) => setTimeout(resolve, this.delayMs)); + const result = await super.write(request); + return { + ...result, + receipt: { externalId: ' ', ts: ` ${deliveredSlackTs(result)} ` }, + }; + } +} + +class ReceiptlessSlackTransport extends IdempotentSlackTransport { + override async write(request: RelayTransportWriteRequest): Promise { + this.attempts.push(request); + return { + path: '/slack/draft.json', + absolutePath: '/slack/draft.json', + receipt: { id: 'mountcmd-without-provider-ts' }, + }; + } +} + +function guardianContext(failWriteCall: number): { ctx: WorkforceCtx; - memoryItems: MemoryItem[]; + files: Map; } { - const memoryItems: MemoryItem[] = []; - let saveCalls = 0; + const files = new Map(); + let writeCalls = 0; const ctx = { + agent: { id: 'sim-agent' }, + deployment: { id: 'sim-deployment' }, persona: { inputs: { SLACK_CHANNEL: 'C0AEKNLDNKW' }, inputSpecs: {}, @@ -81,28 +236,75 @@ function guardianContext(failSaveCall: number): { cwd: '/home/daytona/workspace', readFile: vi.fn(async () => manifest), }, + files: { + read: vi.fn(async (path: string) => { + const contents = files.get(path); + if (contents === undefined) throw new Error(`ENOENT: ${path}`); + return contents; + }), + write: vi.fn(async (path: string, contents: string) => { + writeCalls += 1; + if (writeCalls === failWriteCall) throw new Error('simulated exact-state write failure'); + files.set(path, contents); + }), + }, + credentials: { + tryRequire: vi.fn(() => null), + }, llm: { complete: vi.fn(async () => 'Is this feature working as expected?'), }, memory: { - recall: vi.fn(async () => [...memoryItems]), - save: vi.fn(async (content: string, options?: { tags?: string[]; scope?: string }) => { - saveCalls += 1; - if (saveCalls === failSaveCall) return undefined; - const id = `memory-${saveCalls}`; - memoryItems.push({ - id, - content, - tags: options?.tags ?? [], - scope: 'workspace', - createdAt: new Date(Date.UTC(2026, 6, 18, 0, 0, saveCalls)).toISOString(), - }); - return { id }; + recall: vi.fn(async () => []), + save: vi.fn(async () => undefined), + }, + log: vi.fn(), + } as unknown as WorkforceCtx; + return { ctx, files }; +} + +function exactStateContext( + seed: string, + manifestText = manifest +): { + ctx: WorkforceCtx; + files: Map; +} { + const files = new Map([[cycleStatePath, seed]]); + const ctx = { + agent: { id: 'sim-agent' }, + deployment: { id: 'sim-deployment' }, + persona: { + inputs: { SLACK_CHANNEL: 'C0AEKNLDNKW' }, + inputSpecs: {}, + }, + sandbox: { + cwd: '/home/daytona/workspace', + readFile: vi.fn(async () => manifestText), + }, + files: { + read: vi.fn(async (path: string) => { + const contents = files.get(path); + if (contents === undefined) throw new Error(`ENOENT: ${path}`); + return contents; + }), + write: vi.fn(async (path: string, contents: string) => { + files.set(path, contents); }), }, + credentials: { + tryRequire: vi.fn(() => null), + }, + llm: { + complete: vi.fn(async () => 'Is this feature working as expected?'), + }, + memory: { + recall: vi.fn(async () => []), + save: vi.fn(async () => undefined), + }, log: vi.fn(), } as unknown as WorkforceCtx; - return { ctx, memoryItems }; + return { ctx, files }; } describe('relay-feature-guardian runtime paths', () => { @@ -119,15 +321,14 @@ describe('relay-feature-guardian runtime paths', () => { it('deduplicates an ambiguous post retry and advances after a saved receipt', async () => { const transport = new IdempotentSlackTransport(); const restore = bindPreviewTransport(transport); - const { ctx, memoryItems } = guardianContext(2); + const { ctx, files } = guardianContext(2); try { - // Run 1: the provider delivers Start Broker, then the progress save - // silently returns undefined. Only the pre-post cycle checkpoint remains. + // Run 1: the provider delivers Start Broker, then the exact progress + // checkpoint fails. Only the pre-post empty cycle remains. await guardian.handler(ctx, { type: 'cron.tick' } as never); expect(transport.providerCreates).toBe(1); - expect(memoryItems).toHaveLength(1); - expect(JSON.parse(memoryItems[0].content).checkedIds).toEqual([]); + expect(JSON.parse(files.get(cycleStatePath) ?? '{}').checkedIds).toEqual([]); // Run 2: the same feature uses the same deterministic key, so the // provider receipt is replayed rather than creating a duplicate post. @@ -138,23 +339,23 @@ describe('relay-feature-guardian runtime paths', () => { const retryBody = transport.attempts[1].body as { idempotencyKey: string }; expect(retryBody.idempotencyKey).toBe(firstBody.idempotencyKey); - const completed = JSON.parse(memoryItems.at(-1)?.content ?? '{}') as { + const completed = JSON.parse(files.get(cycleStatePath) ?? '{}') as { checkedIds: string[]; lastPost?: { featureId: string; ts: string }; }; - expect(completed.checkedIds).toEqual(['start-broker']); + expect(completed.checkedIds).toEqual(['broker-up']); expect(completed.lastPost).toEqual({ - featureId: 'start-broker', + featureId: 'broker-up', ts: '1710000001.000100', }); // Run 3: persisted progress selects the next feature in the cycle. await guardian.handler(ctx, { type: 'cron.tick' } as never); expect(transport.providerCreates).toBe(2); - const advanced = JSON.parse(memoryItems.at(-1)?.content ?? '{}') as { + const advanced = JSON.parse(files.get(cycleStatePath) ?? '{}') as { checkedIds: string[]; }; - expect(advanced.checkedIds).toEqual(['start-broker', 'stop-broker']); + expect(advanced.checkedIds).toEqual(['broker-up', 'broker-status']); } finally { restore(); } @@ -163,15 +364,242 @@ describe('relay-feature-guardian runtime paths', () => { it('does not post when the initial cycle checkpoint has no receipt', async () => { const transport = new IdempotentSlackTransport(); const restore = bindPreviewTransport(transport); - const { ctx, memoryItems } = guardianContext(1); + const { ctx, files } = guardianContext(1); try { await guardian.handler(ctx, { type: 'cron.tick' } as never); expect(transport.providerCreates).toBe(0); - expect(memoryItems).toEqual([]); - expect(ctx.log).toHaveBeenCalledWith('error', 'relay-feature-guardian.cycle-checkpoint-failed', { - feature: 'start-broker', + expect(files.size).toBe(0); + expect(ctx.log).toHaveBeenCalledWith( + 'error', + 'relay-feature-guardian.cycle-checkpoint-failed', + expect.objectContaining({ err: expect.stringContaining('simulated exact-state write failure') }) + ); + } finally { + restore(); + } + }); + + it('loads exact cycle state and advances 2/122 then 3/122 across fresh runs', async () => { + const transport = new IdempotentSlackTransport(); + const restore = bindPreviewTransport(transport); + const { ctx, files } = exactStateContext( + JSON.stringify({ + kind: 'relay-feature-guardian:progress', + version: 3, + generation: 1, + checkedIds: ['broker-up'], + cycleStartedAt: '2026-07-18T10:26:47.981Z', + totalFeatures: 122, + lastPost: { featureId: 'broker-up', ts: '1784370419.029509' }, + }) + ); + + try { + await guardian.handler(ctx, { type: 'cron.tick' } as never); + expect(transport.attempts).toHaveLength(1); + expect((transport.attempts[0].body as { text: string }).text).toContain( + 'relay feature check · 2/122 · 120 remaining in cycle' + ); + expect(JSON.parse(files.get(cycleStatePath) ?? '{}').checkedIds).toEqual([ + 'broker-up', + 'broker-status', + ]); + + await guardian.handler(ctx, { type: 'cron.tick' } as never); + expect(transport.attempts).toHaveLength(2); + expect((transport.attempts[1].body as { text: string }).text).toContain( + 'relay feature check · 3/122 · 119 remaining in cycle' + ); + expect(JSON.parse(files.get(cycleStatePath) ?? '{}').checkedIds).toEqual([ + 'broker-up', + 'broker-status', + 'broker-down', + ]); + expect(ctx.memory.recall).not.toHaveBeenCalled(); + expect(ctx.memory.save).not.toHaveBeenCalled(); + } finally { + restore(); + } + }); + + it('preserves progress when the manifest adds features', async () => { + const transport = new IdempotentSlackTransport(); + const restore = bindPreviewTransport(transport); + const staleTotal = { ...progressState(1), totalFeatures: 121 }; + const { ctx, files } = exactStateContext(JSON.stringify(staleTotal)); + try { + await guardian.handler(ctx, { type: 'cron.tick' } as never); + const state = JSON.parse(files.get(CYCLE_STATE_PATH) ?? '{}') as ProgressState; + expect(state.totalFeatures).toBe(122); + expect(state.checkedIds).toEqual(['broker-up', 'broker-status']); + expect(state.lastPost?.featureId).toBe('broker-status'); + } finally { + restore(); + } + }); + + it('reconciles one unchecked feature retirement without resetting the generation', async () => { + const transport = new IdempotentSlackTransport(); + const restore = bindPreviewTransport(transport); + const staleTotal = { ...progressState(1), totalFeatures: 123 }; + const { ctx, files } = exactStateContext(JSON.stringify(staleTotal)); + + try { + await guardian.handler(ctx, { type: 'cron.tick' } as never); + expect(transport.attempts).toHaveLength(1); + expect((transport.attempts[0].body as { text: string }).text).toContain( + 'relay feature check · 2/122 · 120 remaining in cycle' + ); + const state = JSON.parse(files.get(CYCLE_STATE_PATH) ?? '{}') as ProgressState; + expect(state.generation).toBe(1); + expect(state.totalFeatures).toBe(122); + expect(state.checkedIds).toEqual(['broker-up', 'broker-status']); + expect(state.lastPost).toEqual({ + featureId: 'broker-status', + ts: '1710000001.000100', }); + expect(ctx.files.write).toHaveBeenCalledTimes(2); + } finally { + restore(); + } + }); + + it('resets a generation under CAS when the manifest retires a checked feature', async () => { + const transport = new IdempotentSlackTransport(); + const restore = bindPreviewTransport(transport); + const { ctx, files } = exactStateContext( + JSON.stringify({ + kind: 'relay-feature-guardian:progress', + version: 3, + generation: 7, + checkedIds: ['broker-up', 'retired-feature'], + cycleStartedAt: '2026-07-18T10:26:47.981Z', + totalFeatures: 123, + lastPost: { featureId: 'retired-feature', ts: '1784370419.029509' }, + }) + ); + + try { + await guardian.handler(ctx, { type: 'cron.tick' } as never); + expect(transport.attempts).toHaveLength(1); + expect((transport.attempts[0].body as { text: string }).text).toContain( + 'relay feature check · 1/122 · 121 remaining in cycle' + ); + const state = JSON.parse(files.get(CYCLE_STATE_PATH) ?? '{}') as ProgressState; + expect(state.generation).toBe(8); + expect(state.totalFeatures).toBe(122); + expect(state.checkedIds).toEqual(['broker-up']); + expect(state.lastPost).toEqual({ + featureId: 'broker-up', + ts: '1710000001.000100', + }); + } finally { + restore(); + } + }); + + it('preserves exact progress when a suspiciously partial manifest omits checked features', async () => { + const transport = new IdempotentSlackTransport(); + const restore = bindPreviewTransport(transport); + const seed = JSON.stringify({ + kind: 'relay-feature-guardian:progress', + version: 3, + generation: 7, + checkedIds: ['broker-up', 'retired-feature'], + cycleStartedAt: '2026-07-18T10:26:47.981Z', + totalFeatures: 123, + lastPost: { featureId: 'retired-feature', ts: '1784370419.029509' }, + }); + const { ctx, files } = exactStateContext(seed, renderManifest(manifestFeatures.slice(0, 2))); + + try { + await guardian.handler(ctx, { type: 'cron.tick' } as never); + expect(transport.attempts).toHaveLength(0); + expect(JSON.parse(files.get(CYCLE_STATE_PATH) ?? '{}')).toEqual(JSON.parse(seed)); + expect(ctx.log).toHaveBeenCalledWith( + 'error', + 'relay-feature-guardian.progress-reconcile-failed', + expect.objectContaining({ + err: expect.stringContaining('refusing a partial-manifest reset'), + previousTotal: 123, + currentTotal: 2, + }) + ); + } finally { + restore(); + } + }); + + it('preserves exact progress when multiple checked feature IDs disappear at the same total', async () => { + const transport = new IdempotentSlackTransport(); + const restore = bindPreviewTransport(transport); + const seed = JSON.stringify({ + kind: 'relay-feature-guardian:progress', + version: 3, + generation: 7, + checkedIds: ['broker-up', 'retired-feature-a', 'retired-feature-b'], + cycleStartedAt: '2026-07-18T10:26:47.981Z', + totalFeatures: 122, + lastPost: { featureId: 'retired-feature-b', ts: '1784370419.029509' }, + }); + const { ctx, files } = exactStateContext(seed); + + try { + await guardian.handler(ctx, { type: 'cron.tick' } as never); + expect(transport.attempts).toHaveLength(0); + expect(JSON.parse(files.get(CYCLE_STATE_PATH) ?? '{}')).toEqual(JSON.parse(seed)); + expect(ctx.log).toHaveBeenCalledWith( + 'error', + 'relay-feature-guardian.progress-reconcile-failed', + expect.objectContaining({ + err: expect.stringContaining('multiple checked feature ids disappeared'), + previousTotal: 122, + currentTotal: 122, + retiredIds: ['retired-feature-a', 'retired-feature-b'], + }) + ); + } finally { + restore(); + } + }); + + it('preserves exact progress when the manifest read fails', async () => { + const transport = new IdempotentSlackTransport(); + const restore = bindPreviewTransport(transport); + const seed = JSON.stringify(progressState(2)); + const { ctx, files } = exactStateContext(seed); + ctx.sandbox.readFile = vi.fn(async () => { + throw new Error('simulated manifest read failure'); + }); + + try { + await guardian.handler(ctx, { type: 'cron.tick' } as never); + expect(JSON.parse(files.get(CYCLE_STATE_PATH) ?? '{}')).toEqual(JSON.parse(seed)); + expect(ctx.log).toHaveBeenCalledWith( + 'error', + 'relay-feature-guardian.manifest-load-failed', + expect.objectContaining({ err: expect.stringContaining('simulated manifest read failure') }) + ); + } finally { + restore(); + } + }); + + it('fails closed outside invoke simulation when Relayfile credentials are absent', async () => { + const transport = new IdempotentSlackTransport(); + const restore = bindPreviewTransport(transport); + const { ctx } = exactStateContext(JSON.stringify(progressState(1))); + (ctx.agent as { id: string }).id = 'deployed-agent'; + (ctx.deployment as { id: string }).id = 'deployed-run'; + try { + await guardian.handler(ctx, { type: 'cron.tick' } as never); + expect(transport.attempts).toHaveLength(0); + expect(ctx.log).toHaveBeenCalledWith( + 'error', + 'relay-feature-guardian.progress-load-failed', + expect.objectContaining({ err: expect.stringContaining('exact Relayfile credentials') }) + ); } finally { restore(); } @@ -210,5 +638,278 @@ describe('relay-feature-guardian runtime paths', () => { receipt: { externalId: ' ', ts: '1710000002.000200' }, }) ).toBe('1710000002.000200'); + expect( + deliveredSlackTs({ + path: '/invalid-external-id.json', + absolutePath: '/invalid-external-id.json', + receipt: { externalId: 'mountcmd-not-a-ts', ts: '1710000003.000300' }, + }) + ).toBe('1710000003.000300'); + }); +}); + +describe('relay-feature-guardian exact HTTP state', () => { + const credentials = { + url: 'https://relayfile.test', + token: 'relayfile-token', + workspaceId: 'rw_guardian', + }; + + it('bootstraps create-only and updates with the exact loaded revision', async () => { + const server = new RelayfileStateServer(null); + const store = createHttpProgressStore(credentials, { fetchImpl: server.fetch }); + + expect(await store.load(storeFeatures)).toBeNull(); + const initial = await store.save(progressState(0), null, storeFeatures); + const completed = await store.save(progressState(1), initial, storeFeatures); + + expect(completed.state.checkedIds).toEqual(['broker-up']); + expect(server.requests.filter((request) => request.method === 'PUT')).toEqual([ + { method: 'PUT', ifMatch: '0' }, + { method: 'PUT', ifMatch: initial.revision }, + ]); + }); + + it('rejects a stale writer after newer runs checkpoint positions 2 and 3', async () => { + const server = new RelayfileStateServer(progressState(1)); + const store = createHttpProgressStore(credentials, { fetchImpl: server.fetch }); + const staleA = await store.load(storeFeatures); + const runB = await store.load(storeFeatures); + expect(staleA).not.toBeNull(); + expect(runB).not.toBeNull(); + + const position2 = await store.save(progressState(2), runB, storeFeatures); + const runC = await store.load(storeFeatures); + const position3 = await store.save(progressState(3), runC, storeFeatures); + await expect(store.save(progressState(2), staleA, storeFeatures)).rejects.toBeInstanceOf( + ProgressStateConflictError + ); + + expect(position2.state.lastPost?.featureId).toBe('broker-status'); + expect(position3.state.lastPost?.featureId).toBe('broker-down'); + expect(server.state()).toEqual(progressState(3)); + }); + + it('uses the loaded revision when resetting a genuinely retired feature', async () => { + const retiredState: ProgressState = { + kind: 'relay-feature-guardian:progress', + version: 3, + generation: 7, + checkedIds: ['broker-up', 'retired-feature'], + cycleStartedAt: '2026-07-18T10:26:47.981Z', + totalFeatures: 123, + lastPost: { featureId: 'retired-feature', ts: '1784370419.029509' }, + }; + const server = new RelayfileStateServer(retiredState); + const store = createHttpProgressStore(credentials, { fetchImpl: server.fetch }); + const loaded = await store.load(storeFeatures); + expect(loaded).not.toBeNull(); + + const reset: ProgressState = { + kind: 'relay-feature-guardian:progress', + version: 3, + generation: 8, + checkedIds: [], + cycleStartedAt: '2026-07-18T11:26:47.981Z', + totalFeatures: 122, + }; + const saved = await store.save(reset, loaded, storeFeatures); + + expect(saved.state).toEqual(reset); + expect(server.requests.filter((request) => request.method === 'PUT')).toEqual([ + { method: 'PUT', ifMatch: loaded?.revision ?? '' }, + ]); + }); + + it('uses the loaded revision for a bounded downward total reconciliation', async () => { + const previousState = { ...progressState(1), totalFeatures: 123 }; + const server = new RelayfileStateServer(previousState); + const store = createHttpProgressStore(credentials, { fetchImpl: server.fetch }); + const loaded = await store.load(storeFeatures); + expect(loaded).not.toBeNull(); + + const reconciled = { ...previousState, totalFeatures: 122 }; + const saved = await store.save(reconciled, loaded, storeFeatures); + + expect(saved.state).toEqual(reconciled); + expect(server.requests.filter((request) => request.method === 'PUT')).toEqual([ + { method: 'PUT', ifMatch: loaded?.revision ?? '' }, + ]); + }); + + it('does not let an old completed generation overwrite its CAS reset', async () => { + const server = new RelayfileStateServer(progressState(122)); + const store = createHttpProgressStore(credentials, { fetchImpl: server.fetch }); + const completedGeneration = await store.load(storeFeatures); + expect(completedGeneration).not.toBeNull(); + + const reset = await store.save(progressState(0, 2), completedGeneration, storeFeatures); + await expect(store.save(progressState(122), completedGeneration, storeFeatures)).rejects.toBeInstanceOf( + ProgressStateConflictError + ); + + expect(reset.state.generation).toBe(2); + expect(server.state()).toEqual(progressState(0, 2)); + }); + + it('treats only 404 as absent and fails closed on authorization errors', async () => { + const absent = new RelayfileStateServer(null); + const absentStore = createHttpProgressStore(credentials, { fetchImpl: absent.fetch }); + expect(await absentStore.load(storeFeatures)).toBeNull(); + + const forbidden = new RelayfileStateServer(progressState(1)); + forbidden.failGetStatus = 403; + const forbiddenStore = createHttpProgressStore(credentials, { fetchImpl: forbidden.fetch }); + await expect(forbiddenStore.load(storeFeatures)).rejects.toThrow('HTTP 403'); + }); + + it('bounds a hung GET and a hung PUT with one state transaction deadline', async () => { + vi.useFakeTimers(); + try { + const hungGet = new RelayfileStateServer(progressState(1)); + hungGet.hangMethod = 'GET'; + const getStore = createHttpProgressStore(credentials, { + fetchImpl: hungGet.fetch, + timeoutMs: 25, + }); + const getResult = getStore.load(storeFeatures); + const getRejection = expect(getResult).rejects.toThrow('cycle state load timed out after 25ms'); + await vi.advanceTimersByTimeAsync(26); + await getRejection; + + const hungPut = new RelayfileStateServer(null); + hungPut.hangMethod = 'PUT'; + const putStore = createHttpProgressStore(credentials, { + fetchImpl: hungPut.fetch, + timeoutMs: 25, + }); + const putResult = putStore.save(progressState(0), null, storeFeatures); + const putRejection = expect(putResult).rejects.toThrow('cycle state save timed out after 25ms'); + await vi.advanceTimersByTimeAsync(26); + await putRejection; + } finally { + vi.useRealTimers(); + } + }); + + it('fails closed when the exact GET read-back does not match the PUT', async () => { + const server = new RelayfileStateServer(null); + server.corruptReadBack = true; + const store = createHttpProgressStore(credentials, { fetchImpl: server.fetch }); + await expect(store.save(progressState(0), null, storeFeatures)).rejects.toThrow( + 'read-back did not match' + ); + }); + + it('rejects oversized UTF-8 state before issuing a PUT', async () => { + const featureIds = Array.from({ length: 300 }, (_, index) => `feature-${index}-${'x'.repeat(220)}`); + const features = featureIds.map((id, index) => ({ + id, + name: `Feature ${index}`, + cli: `relay feature-${index}`, + desc: `Checks feature ${index}.`, + tier: 1, + criticality: 'critical' as const, + })); + const state = (checkedIds: string[]): ProgressState => ({ + kind: 'relay-feature-guardian:progress', + version: 3, + generation: 1, + checkedIds, + cycleStartedAt: '2026-07-18T10:26:47.981Z', + totalFeatures: features.length, + ...(checkedIds.length > 0 + ? { + lastPost: { + featureId: checkedIds.at(-1) as string, + ts: '1784370419.029509', + }, + } + : {}), + }); + const previousState = state(featureIds.slice(0, -1)); + const server = new RelayfileStateServer(null); + const store = createHttpProgressStore(credentials, { fetchImpl: server.fetch }); + + await expect( + Promise.resolve().then(() => + store.save(state(featureIds), { state: previousState, revision: '7' }, features) + ) + ).rejects.toThrow('cycle state exceeds size limit'); + expect(server.requests).toEqual([]); + }); + + it.each([ + ['duplicate ids', { ...progressState(2), checkedIds: ['broker-up', 'broker-up'] }], + ['malformed historical id', { ...progressState(1), checkedIds: [''] }], + ['invalid ts', { ...progressState(1), lastPost: { featureId: 'broker-up', ts: 'not-a-slack-ts' } }], + ['invalid cycle time', { ...progressState(1), cycleStartedAt: 'yesterday' }], + ])('rejects bounded v3 state with %s', async (_label, invalid) => { + const server = new RelayfileStateServer(null); + server.content = `${JSON.stringify(invalid)}\n`; + const store = createHttpProgressStore(credentials, { fetchImpl: server.fetch }); + await expect(store.load(storeFeatures)).rejects.toThrow(/cycle state/); + }); +}); + +describe('relay-feature-guardian delayed Slack receipts', () => { + it('passes the production receipt deadline and poll interval to the Slack helper', async () => { + const transport = new IdempotentSlackTransport(); + const restore = bindPreviewTransport(transport); + const { ctx } = exactStateContext(JSON.stringify(progressState(1))); + const observedOptions: unknown[] = []; + const createSlackClient = ((options?: Parameters[0]) => { + observedOptions.push(options); + return slackClient(options); + }) as typeof slackClient; + + try { + await runGuardian(ctx, { type: 'cron.tick' } as never, { createSlackClient }); + expect(observedOptions).toContainEqual({ + writebackTimeoutMs: 15_000, + writebackPollMs: 250, + }); + } finally { + restore(); + } + }); + + it('waits beyond the old 3s window and checkpoints only a real trimmed ts', async () => { + expect(SLACK_WRITEBACK_TIMEOUT_MS).toBe(15_000); + expect(SLACK_WRITEBACK_POLL_MS).toBe(250); + vi.useFakeTimers(); + const transport = new DelayedSlackTransport(3_500); + const restore = bindPreviewTransport(transport); + const { ctx, files } = exactStateContext(JSON.stringify(progressState(1))); + try { + const run = guardian.handler(ctx, { type: 'cron.tick' } as never); + await vi.advanceTimersByTimeAsync(3_001); + expect(JSON.parse(files.get(CYCLE_STATE_PATH) ?? '{}').checkedIds).toEqual(['broker-up']); + await vi.advanceTimersByTimeAsync(500); + await run; + + const state = JSON.parse(files.get(CYCLE_STATE_PATH) ?? '{}') as ProgressState; + expect(state.checkedIds).toEqual(['broker-up', 'broker-status']); + expect(state.lastPost?.ts).toBe('1710000001.000100'); + } finally { + restore(); + vi.useRealTimers(); + } + }); + + it('never advances on a receipt-shaped draft without provider ts', async () => { + const transport = new ReceiptlessSlackTransport(); + const restore = bindPreviewTransport(transport); + const { ctx, files } = exactStateContext(JSON.stringify(progressState(1))); + try { + await guardian.handler(ctx, { type: 'cron.tick' } as never); + expect(JSON.parse(files.get(CYCLE_STATE_PATH) ?? '{}').checkedIds).toEqual(['broker-up']); + expect(ctx.log).toHaveBeenCalledWith('error', 'relay-feature-guardian.post-failed', { + channel: 'C0AEKNLDNKW', + feature: 'broker-status', + }); + } finally { + restore(); + } }); }); diff --git a/.agentworkforce/agents/relay-feature-guardian/agent.ts b/.agentworkforce/agents/relay-feature-guardian/agent.ts index dac5910c4..4e3d5e323 100644 --- a/.agentworkforce/agents/relay-feature-guardian/agent.ts +++ b/.agentworkforce/agents/relay-feature-guardian/agent.ts @@ -3,7 +3,7 @@ * * Hourly cron tick: * 1. Read the feature list from the cloned relay repository - * 2. Load feature progress from memory + * 2. Load feature progress from an exact, revisioned Relayfile record * 3. Pick the next unchecked feature (ordered by criticality then tier) * 4. Generate a concise quiz question via ctx.llm * 5. Post to Slack with @mentions for Will and Khaliq @@ -11,9 +11,15 @@ * * After the full manifest is covered, the cycle resets. */ -import { defineAgent, type WorkforceCtx } from '@agentworkforce/runtime'; +import { + defineAgent, + type RelayfileCredentials, + type WorkforceCtx, + type WorkforceEvent, +} from '@agentworkforce/runtime'; import { input } from '@agentworkforce/delivery'; import { slackClient, type WritebackResult } from '@relayfile/relay-helpers'; +import { randomUUID } from 'node:crypto'; import { parse } from 'yaml'; // ── manifest types ──────────────────────────────────────────────────────────── @@ -90,9 +96,10 @@ async function loadFeatures(ctx: WorkforceCtx): Promise { // ── progress tracking ───────────────────────────────────────────────────────── -interface ProgressState { +export interface ProgressState { kind: 'relay-feature-guardian:progress'; - version: 1 | 2; + version: 3; + generation: number; checkedIds: string[]; cycleStartedAt: string; totalFeatures: number; @@ -102,36 +109,411 @@ interface ProgressState { }; } -async function loadProgress(ctx: WorkforceCtx): Promise { - const items = await ctx.memory.recall('relay-feature-guardian cycle progress', { - tags: ['relay-feature-guardian:progress'], - scope: 'workspace', - limit: 5, +export const CYCLE_STATE_PATH = '/memory/workspace/relay-feature-guardian/cycle-state.json'; +export const STATE_IO_TIMEOUT_MS = 5_000; +export const SLACK_WRITEBACK_TIMEOUT_MS = 15_000; +export const SLACK_WRITEBACK_POLL_MS = 250; + +const MAX_STATE_BYTES = 64 * 1024; +const MAX_SAFE_MANIFEST_SHRINK = 1; +const UTF8_ENCODER = new TextEncoder(); + +export interface ProgressSnapshot { + state: ProgressState; + revision: string; +} + +export interface ProgressStore { + load(features: Feature[]): Promise; + save( + state: ProgressState, + expected: ProgressSnapshot | null, + features: Feature[] + ): Promise; +} + +type FetchLike = typeof fetch; + +export class ProgressStateConflictError extends Error { + constructor(message = 'relay-feature-guardian cycle state revision conflict') { + super(message); + this.name = 'ProgressStateConflictError'; + } +} + +function isRecord(value: unknown): value is Record { + return value !== null && typeof value === 'object' && !Array.isArray(value); +} + +function isCanonicalIsoTimestamp(value: unknown): value is string { + if (typeof value !== 'string') return false; + const parsed = new Date(value); + return Number.isFinite(parsed.valueOf()) && parsed.toISOString() === value; +} + +function isSlackTs(value: unknown): value is string { + return typeof value === 'string' && /^\d+\.\d+$/.test(value.trim()); +} + +function assertStateSize(content: string): void { + if (UTF8_ENCODER.encode(content).byteLength > MAX_STATE_BYTES) { + throw new Error('cycle state exceeds size limit'); + } +} + +function parseProgressState( + value: unknown, + features: Feature[], + options: { allowHistoricalIds?: boolean } = {} +): ProgressState { + if (!isRecord(value)) throw new Error('cycle state must be an object'); + if (value.kind !== 'relay-feature-guardian:progress' || value.version !== 3) { + throw new Error('cycle state kind/version is invalid'); + } + if (!Number.isSafeInteger(value.generation) || (value.generation as number) < 1) { + throw new Error('cycle state generation is invalid'); + } + if (!isCanonicalIsoTimestamp(value.cycleStartedAt)) { + throw new Error('cycle state cycleStartedAt is invalid'); + } + if ( + !Number.isSafeInteger(value.totalFeatures) || + (value.totalFeatures as number) < 1 || + (value.totalFeatures as number) > 10_000 + ) { + throw new Error('cycle state totalFeatures is invalid'); + } + if (!Array.isArray(value.checkedIds) || value.checkedIds.length > (value.totalFeatures as number)) { + throw new Error('cycle state checkedIds is invalid'); + } + + const knownIds = new Set(features.map((feature) => feature.id)); + const checkedIds = value.checkedIds.map((id) => { + if ( + typeof id !== 'string' || + id.length === 0 || + id.length > 256 || + (!options.allowHistoricalIds && !knownIds.has(id)) + ) { + throw new Error('cycle state contains an unknown feature id'); + } + return id; }); - for (const item of [...items].sort((a, b) => (b.createdAt ?? '').localeCompare(a.createdAt ?? ''))) { - try { - const parsed = JSON.parse(item.content) as unknown; - if ( - parsed && - typeof parsed === 'object' && - (parsed as ProgressState).kind === 'relay-feature-guardian:progress' - ) { - return parsed as ProgressState; + if (new Set(checkedIds).size !== checkedIds.length) { + throw new Error('cycle state contains duplicate feature ids'); + } + + let lastPost: ProgressState['lastPost']; + if (value.lastPost !== undefined) { + if ( + !isRecord(value.lastPost) || + typeof value.lastPost.featureId !== 'string' || + !checkedIds.includes(value.lastPost.featureId) || + !isSlackTs(value.lastPost.ts) + ) { + throw new Error('cycle state lastPost is invalid'); + } + lastPost = { + featureId: value.lastPost.featureId, + ts: value.lastPost.ts.trim(), + }; + } + if (checkedIds.length > 0 && !lastPost) { + throw new Error('cycle state with progress requires lastPost'); + } + if (lastPost && lastPost.featureId !== checkedIds.at(-1)) { + throw new Error('cycle state lastPost must describe the latest checked feature'); + } + + return { + kind: 'relay-feature-guardian:progress', + version: 3, + generation: value.generation as number, + checkedIds, + cycleStartedAt: value.cycleStartedAt, + totalFeatures: value.totalFeatures as number, + ...(lastPost ? { lastPost } : {}), + }; +} + +function retiredFeatureIds(state: ProgressState, features: Feature[]): string[] { + const currentIds = new Set(features.map((feature) => feature.id)); + return state.checkedIds.filter((id) => !currentIds.has(id)); +} + +type ManifestDelta = { + kind: 'preserve' | 'reset-checked-retirement' | 'unsafe'; + retiredIds: string[]; + removedCount: number; + reason?: string; +}; + +/** + * Complete manifest-delta matrix: + * - additions and one unchecked removal preserve the current generation; + * - one checked retirement/rename resets the generation under CAS; + * - larger shrink or multiple missing checked IDs is suspicious and fail-closed; + * - manifest read failures never reach this classifier. + */ +function classifyManifestDelta(state: ProgressState, features: Feature[]): ManifestDelta { + const removedCount = Math.max(0, state.totalFeatures - features.length); + const retiredIds = retiredFeatureIds(state, features); + if (removedCount > MAX_SAFE_MANIFEST_SHRINK) { + return { + kind: 'unsafe', + retiredIds, + removedCount, + reason: 'manifest feature count shrank by more than one; refusing a partial-manifest reset', + }; + } + if (retiredIds.length > 1) { + return { + kind: 'unsafe', + retiredIds, + removedCount, + reason: 'multiple checked feature ids disappeared; refusing an ambiguous manifest reset', + }; + } + return { + kind: retiredIds.length === 1 ? 'reset-checked-retirement' : 'preserve', + retiredIds, + removedCount, + }; +} + +function assertValidTransition( + previous: ProgressSnapshot | null, + next: ProgressState, + features: Feature[] +): void { + if (!previous) { + if (next.generation !== 1 || next.checkedIds.length !== 0 || next.lastPost) { + throw new Error('new cycle state must bootstrap an empty generation 1'); + } + return; + } + + const prior = previous.state; + if (next.generation === prior.generation) { + if (next.cycleStartedAt !== prior.cycleStartedAt) { + throw new Error('cycleStartedAt is immutable within a generation'); + } + if ( + next.checkedIds.length < prior.checkedIds.length || + next.checkedIds.length > prior.checkedIds.length + 1 || + !prior.checkedIds.every((id, index) => next.checkedIds[index] === id) + ) { + throw new Error('cycle progress cannot regress or skip a checkpoint'); + } + if (next.checkedIds.length === prior.checkedIds.length) { + if (JSON.stringify(next.lastPost) !== JSON.stringify(prior.lastPost)) { + throw new Error('reconciliation cannot overwrite lastPost'); } - } catch { - // skip malformed + } else if (next.lastPost?.featureId !== next.checkedIds.at(-1)) { + throw new Error('new progress must checkpoint the newly checked feature'); } + return; + } + + const manifestDelta = classifyManifestDelta(prior, features); + const retirementResetAllowed = manifestDelta.kind === 'reset-checked-retirement'; + const allCurrentFeaturesChecked = features.every((feature) => prior.checkedIds.includes(feature.id)); + if ( + next.generation !== prior.generation + 1 || + (!allCurrentFeaturesChecked && !retirementResetAllowed) || + next.checkedIds.length !== 0 || + next.lastPost || + new Date(next.cycleStartedAt) <= new Date(prior.cycleStartedAt) + ) { + throw new Error('cycle generation reset is invalid'); } - return null; } -async function saveProgress(ctx: WorkforceCtx, state: ProgressState): Promise { - const receipt = await ctx.memory.save(JSON.stringify(state), { - tags: ['relay-feature-guardian:progress'], - scope: 'workspace', - ttlSeconds: 60 * 60 * 24 * 14, // 14 days +async function withDeadline( + label: string, + timeoutMs: number, + run: (signal: AbortSignal) => Promise +): Promise { + const controller = new AbortController(); + let timeout: ReturnType | undefined; + const expired = new Promise((_, reject) => { + timeout = setTimeout(() => { + controller.abort(); + reject(new Error(`${label} timed out after ${timeoutMs}ms`)); + }, timeoutMs); }); - return receipt ? receipt.id : null; + try { + return await Promise.race([run(controller.signal), expired]); + } finally { + if (timeout !== undefined) clearTimeout(timeout); + } +} + +function normalizeEtag(value: string | null): string { + return (value ?? '').trim().replace(/^W\//, '').replace(/^"|"$/g, ''); +} + +function relayfileStateUrl(credentials: RelayfileCredentials): URL { + const url = new URL( + `/v1/workspaces/${encodeURIComponent(credentials.workspaceId)}/fs/file`, + `${credentials.url.replace(/\/+$/, '')}/` + ); + url.searchParams.set('path', CYCLE_STATE_PATH); + return url; +} + +function requestHeaders(credentials: RelayfileCredentials, correlationId: string): HeadersInit { + return { + Authorization: `Bearer ${credentials.token}`, + 'X-Correlation-Id': correlationId, + }; +} + +async function readHttpSnapshot( + credentials: RelayfileCredentials, + features: Feature[], + fetchImpl: FetchLike, + signal: AbortSignal, + correlationId: string, + allowHistoricalIds = false +): Promise { + const response = await fetchImpl(relayfileStateUrl(credentials), { + method: 'GET', + headers: requestHeaders(credentials, correlationId), + signal, + }); + if (response.status === 404) return null; + if (!response.ok) throw new Error(`cycle state GET failed with HTTP ${response.status}`); + + const file = (await response.json()) as unknown; + if (!isRecord(file) || file.path !== CYCLE_STATE_PATH || typeof file.content !== 'string') { + throw new Error('cycle state GET returned an invalid file'); + } + assertStateSize(file.content); + const bodyRevision = typeof file.revision === 'string' ? file.revision.trim() : ''; + const etagRevision = normalizeEtag(response.headers.get('etag')); + const revision = bodyRevision || etagRevision; + if (!revision || (bodyRevision && etagRevision && bodyRevision !== etagRevision)) { + throw new Error('cycle state GET returned an invalid revision'); + } + + let parsed: unknown; + try { + parsed = JSON.parse(file.content); + } catch { + throw new Error('cycle state contains invalid JSON'); + } + return { + state: parseProgressState(parsed, features, { allowHistoricalIds }), + revision, + }; +} + +export function createHttpProgressStore( + credentials: RelayfileCredentials, + options: { fetchImpl?: FetchLike; timeoutMs?: number } = {} +): ProgressStore { + const fetchImpl = options.fetchImpl ?? fetch; + const timeoutMs = options.timeoutMs ?? STATE_IO_TIMEOUT_MS; + return { + load: (features) => + withDeadline('cycle state load', timeoutMs, (signal) => + readHttpSnapshot( + credentials, + features, + fetchImpl, + signal, + `guardian-state-load-${randomUUID()}`, + true + ) + ), + save: (state, expected, features) => { + const canonical = parseProgressState(state, features); + if (canonical.totalFeatures !== features.length) { + throw new Error('cycle state totalFeatures must match the current manifest'); + } + const content = `${JSON.stringify(canonical)}\n`; + assertStateSize(content); + assertValidTransition(expected, canonical, features); + return withDeadline('cycle state save', timeoutMs, async (signal) => { + const correlationId = `guardian-state-save-${randomUUID()}`; + const response = await fetchImpl(relayfileStateUrl(credentials), { + method: 'PUT', + headers: { + ...requestHeaders(credentials, correlationId), + 'Content-Type': 'application/octet-stream', + 'X-Relayfile-Encoding': 'utf-8', + 'X-Relayfile-Content-Type': 'application/json', + 'X-Workspace-Id': credentials.workspaceId, + 'If-Match': expected?.revision ?? '0', + }, + body: content, + signal, + }); + if (response.status === 409) throw new ProgressStateConflictError(); + if (!response.ok) throw new Error(`cycle state PUT failed with HTTP ${response.status}`); + + const readBack = await readHttpSnapshot(credentials, features, fetchImpl, signal, correlationId); + if (!readBack || JSON.stringify(readBack.state) !== JSON.stringify(canonical)) { + throw new Error('cycle state read-back did not match the saved state'); + } + if (expected && readBack.revision === expected.revision) { + throw new Error('cycle state revision did not advance'); + } + return readBack; + }); + }, + }; +} + +function previewRevision(state: ProgressState): string { + return `preview:${JSON.stringify(state)}`; +} + +function createPreviewProgressStore(ctx: WorkforceCtx): ProgressStore { + const read = async (features: Feature[], allowHistoricalIds = false): Promise => { + let content: string; + try { + content = await ctx.files.read(CYCLE_STATE_PATH); + } catch (error) { + if (String(error).includes('ENOENT')) return null; + throw error; + } + assertStateSize(content); + const state = parseProgressState(JSON.parse(content) as unknown, features, { + allowHistoricalIds, + }); + return { state, revision: previewRevision(state) }; + }; + return { + load: (features) => read(features, true), + save: async (state, expected, features) => { + const canonical = parseProgressState(state, features); + if (canonical.totalFeatures !== features.length) { + throw new Error('cycle state totalFeatures must match the current manifest'); + } + const content = `${JSON.stringify(canonical)}\n`; + assertStateSize(content); + assertValidTransition(expected, canonical, features); + const current = await read(features, true); + if (current?.revision !== expected?.revision) throw new ProgressStateConflictError(); + await ctx.files.write(CYCLE_STATE_PATH, content); + const readBack = await read(features); + if (!readBack || JSON.stringify(readBack.state) !== JSON.stringify(canonical)) { + throw new Error('cycle state read-back did not match the saved state'); + } + return readBack; + }, + }; +} + +function createProgressStore(ctx: WorkforceCtx): ProgressStore { + const credentials = ctx.credentials.tryRequire(); + if (credentials) return createHttpProgressStore(credentials.relayfile); + if (ctx.agent.id === 'sim-agent' && ctx.deployment.id === 'sim-deployment') { + return createPreviewProgressStore(ctx); + } + throw new Error('exact Relayfile credentials are required for guardian cycle state'); } export function featurePostIdempotencyKey(cycleStartedAt: string, featureId: string): string { @@ -141,8 +523,9 @@ export function featurePostIdempotencyKey(cycleStartedAt: string, featureId: str export function deliveredSlackTs(result: WritebackResult | null | undefined): string { const receipt = result?.receipt as { externalId?: unknown; ts?: unknown } | undefined; const externalId = typeof receipt?.externalId === 'string' ? receipt.externalId.trim() : ''; - if (externalId) return externalId; - return typeof receipt?.ts === 'string' ? receipt.ts.trim() : ''; + if (isSlackTs(externalId)) return externalId; + const ts = typeof receipt?.ts === 'string' ? receipt.ts.trim() : ''; + return isSlackTs(ts) ? ts : ''; } // ── feature selection ───────────────────────────────────────────────────────── @@ -204,135 +587,240 @@ async function generateQuizMessage(ctx: WorkforceCtx, feature: Feature): Promise // ── agent definition ────────────────────────────────────────────────────────── -export default defineAgent({ - schedules: [{ name: 'hourly-check', cron: '0 * * * *', tz: 'America/New_York' }], - handler: async (ctx, _event) => { - const channel = input(ctx, 'SLACK_CHANNEL'); - if (!channel) { - ctx.log('warn', 'relay-feature-guardian.no-channel', { reason: 'SLACK_CHANNEL not configured' }); +export interface GuardianDependencies { + createSlackClient?: typeof slackClient; +} + +export async function runGuardian( + ctx: WorkforceCtx, + _event: WorkforceEvent, + dependencies: GuardianDependencies = {} +): Promise { + const createSlackClient = dependencies.createSlackClient ?? slackClient; + const channel = input(ctx, 'SLACK_CHANNEL'); + if (!channel) { + ctx.log('warn', 'relay-feature-guardian.no-channel', { reason: 'SLACK_CHANNEL not configured' }); + return; + } + + // Load the live feature list from the manifest + let features: Feature[]; + try { + features = await loadFeatures(ctx); + } catch (err) { + const absPath = resolveManifestPath(ctx.sandbox.cwd); + ctx.log('error', 'relay-feature-guardian.manifest-load-failed', { path: absPath, err: String(err) }); + const isNotFound = String(err).includes('ENOENT'); + const errMsg = isNotFound + ? `⚠️ *relay-feature-guardian* can't find the feature manifest in the cloned relay repository at \`${RELAY_REPO_RELPATH}/${MANIFEST_RELPATH}\`.` + : `⚠️ *relay-feature-guardian* failed to load the feature manifest: \`${String(err)}\``; + await createSlackClient() + .post(channel, errMsg) + .catch(() => undefined); + return; + } + ctx.log('info', 'relay-feature-guardian.manifest-loaded', { + path: resolveManifestPath(ctx.sandbox.cwd), + features: features.length, + }); + if (features.length === 0) { + ctx.log('error', 'relay-feature-guardian.no-features', { reason: 'manifest parsed but empty' }); + await createSlackClient() + .post( + channel, + '⚠️ *relay-feature-guardian* loaded the manifest but found no features. Check `.agentworkforce/features/manifest.yaml`.' + ) + .catch(() => undefined); + return; + } + + const totalFeatures = features.length; + let store: ProgressStore; + let progress: ProgressSnapshot | null; + try { + store = createProgressStore(ctx); + progress = await store.load(features); + } catch (err) { + ctx.log('error', 'relay-feature-guardian.progress-load-failed', { err: String(err) }); + return; + } + + // A missing exact record is the only bootstrap case. Persist and read back + // the empty generation before any Slack side effect. + if (!progress) { + const initial: ProgressState = { + kind: 'relay-feature-guardian:progress', + version: 3, + generation: 1, + checkedIds: [], + cycleStartedAt: new Date().toISOString(), + totalFeatures, + }; + try { + progress = await store.save(initial, null, features); + } catch (err) { + ctx.log('error', 'relay-feature-guardian.cycle-checkpoint-failed', { err: String(err) }); return; } + } + + const manifestDelta = classifyManifestDelta(progress.state, features); + const { retiredIds } = manifestDelta; + if (manifestDelta.kind === 'unsafe') { + ctx.log('error', 'relay-feature-guardian.progress-reconcile-failed', { + err: manifestDelta.reason, + previousTotal: progress.state.totalFeatures, + currentTotal: totalFeatures, + retiredIds, + }); + return; + } - // Load the live feature list from the manifest - let features: Feature[]; + // A successfully parsed manifest can retire a feature mid-cycle. Historical + // IDs are accepted only at load; reset them under exact-revision CAS before + // any Slack side effect so the persisted state is canonical again. + if (manifestDelta.kind === 'reset-checked-retirement') { + const previousStart = new Date(progress.state.cycleStartedAt).valueOf(); + const reset: ProgressState = { + kind: 'relay-feature-guardian:progress', + version: 3, + generation: progress.state.generation + 1, + checkedIds: [], + cycleStartedAt: new Date(Math.max(Date.now(), previousStart + 1)).toISOString(), + totalFeatures, + }; try { - features = await loadFeatures(ctx); + progress = await store.save(reset, progress, features); + ctx.log('info', 'relay-feature-guardian.manifest-retirement-reset', { + retiredIds, + generation: progress.state.generation, + total: totalFeatures, + }); } catch (err) { - const absPath = resolveManifestPath(ctx.sandbox.cwd); - ctx.log('error', 'relay-feature-guardian.manifest-load-failed', { path: absPath, err: String(err) }); - const isNotFound = String(err).includes('ENOENT'); - const errMsg = isNotFound - ? `⚠️ *relay-feature-guardian* can't find the feature manifest in the cloned relay repository at \`${RELAY_REPO_RELPATH}/${MANIFEST_RELPATH}\`.` - : `⚠️ *relay-feature-guardian* failed to load the feature manifest: \`${String(err)}\``; - await slackClient() - .post(channel, errMsg) - .catch(() => undefined); + ctx.log('error', 'relay-feature-guardian.progress-reconcile-failed', { err: String(err) }); return; } - ctx.log('info', 'relay-feature-guardian.manifest-loaded', { - path: resolveManifestPath(ctx.sandbox.cwd), - features: features.length, - }); - if (features.length === 0) { - ctx.log('error', 'relay-feature-guardian.no-features', { reason: 'manifest parsed but empty' }); - await slackClient() - .post( - channel, - '⚠️ *relay-feature-guardian* loaded the manifest but found no features. Check `.agentworkforce/features/manifest.yaml`.' - ) - .catch(() => undefined); + } else if (progress.state.totalFeatures !== totalFeatures) { + // Manifest additions change the denominator but must never discard already + // checked feature ids or overwrite the last delivered receipt. + const reconciled: ProgressState = { + ...progress.state, + totalFeatures, + }; + try { + progress = await store.save(reconciled, progress, features); + } catch (err) { + ctx.log('error', 'relay-feature-guardian.progress-reconcile-failed', { err: String(err) }); return; } + } - // Load progress or bootstrap a fresh cycle - const progress = await loadProgress(ctx); - const checkedIds = new Set(progress?.checkedIds ?? []); - const totalFeatures = features.length; - let cycleStartedAt = progress?.cycleStartedAt ?? new Date().toISOString(); - let needsCycleCheckpoint = !progress; - - // Pick the next unchecked feature; reset if the cycle is complete - let feature = pickNextFeature(features, checkedIds); - if (!feature) { - ctx.log('info', 'relay-feature-guardian.cycle-complete', { total: totalFeatures }); - checkedIds.clear(); - cycleStartedAt = new Date().toISOString(); - needsCycleCheckpoint = true; - feature = pickNextFeature(features, checkedIds); - } - if (!feature) return; - - // Persist a stable cycle identity before the side effect. If memory is - // unavailable, stop instead of posting a feature we cannot checkpoint. - if (needsCycleCheckpoint) { - const checkpointId = await saveProgress(ctx, { - kind: 'relay-feature-guardian:progress', - version: 2, - checkedIds: [...checkedIds], - cycleStartedAt, - totalFeatures, - }); - if (!checkpointId) { - ctx.log('error', 'relay-feature-guardian.cycle-checkpoint-failed', { - feature: feature.id, - }); - return; - } + let checkedIds = new Set(progress.state.checkedIds); + + // Pick the next unchecked feature; reset if the cycle is complete + let feature = pickNextFeature(features, checkedIds); + if (!feature) { + ctx.log('info', 'relay-feature-guardian.cycle-complete', { total: totalFeatures }); + const previousStart = new Date(progress.state.cycleStartedAt).valueOf(); + const reset: ProgressState = { + kind: 'relay-feature-guardian:progress', + version: 3, + generation: progress.state.generation + 1, + checkedIds: [], + cycleStartedAt: new Date(Math.max(Date.now(), previousStart + 1)).toISOString(), + totalFeatures, + }; + try { + progress = await store.save(reset, progress, features); + } catch (err) { + ctx.log('error', 'relay-feature-guardian.cycle-checkpoint-failed', { err: String(err) }); + return; } + checkedIds = new Set(); + feature = pickNextFeature(features, checkedIds); + } + if (!feature) return; + + // Build @mention string + const userWill = input(ctx, 'SLACK_USER_WILL'); + const userKhaliq = input(ctx, 'SLACK_USER_KHALIQ'); + const mentions = [userWill && `<@${userWill}>`, userKhaliq && `<@${userKhaliq}>`].filter(Boolean).join(' '); + const mentionPrefix = mentions ? `${mentions} — ` : ''; + + // Generate quiz message + const quizBody = await generateQuizMessage(ctx, feature); + const remaining = totalFeatures - checkedIds.size - 1; + const progressNote = `_[relay feature check · ${checkedIds.size + 1}/${totalFeatures} · ${remaining} remaining in cycle]_`; + const message = [mentionPrefix + quizBody, '', progressNote].join('\n'); + ctx.log('info', 'relay-feature-guardian.posting', { + channel, + feature: feature.id, + index: checkedIds.size + 1, + total: totalFeatures, + remaining, + }); - // Build @mention string - const userWill = input(ctx, 'SLACK_USER_WILL'); - const userKhaliq = input(ctx, 'SLACK_USER_KHALIQ'); - const mentions = [userWill && `<@${userWill}>`, userKhaliq && `<@${userKhaliq}>`] - .filter(Boolean) - .join(' '); - const mentionPrefix = mentions ? `${mentions} — ` : ''; - - // Generate quiz message - const quizBody = await generateQuizMessage(ctx, feature); - const remaining = totalFeatures - checkedIds.size - 1; - const progressNote = `_[relay feature check · ${checkedIds.size + 1}/${totalFeatures} · ${remaining} remaining in cycle]_`; - const message = [mentionPrefix + quizBody, '', progressNote].join('\n'); - - // Post to Slack - const slack = slackClient(); - const result = await slack.messages.write( + // Post to Slack + const slack = createSlackClient({ + writebackTimeoutMs: SLACK_WRITEBACK_TIMEOUT_MS, + writebackPollMs: SLACK_WRITEBACK_POLL_MS, + }); + let result: WritebackResult; + try { + result = await slack.messages.write( { channelId: channel }, { text: message, - idempotencyKey: featurePostIdempotencyKey(cycleStartedAt, feature.id), + idempotencyKey: featurePostIdempotencyKey(progress.state.cycleStartedAt, feature.id), } ); - const ts = deliveredSlackTs(result); - if (!ts) { - ctx.log('error', 'relay-feature-guardian.post-failed', { channel, feature: feature.id }); - return; - } - - // Checkpoint immediately after the confirmed provider receipt. The stable - // idempotency key makes a retry safe if this save times out or the run caps. - checkedIds.add(feature.id); - const checkpointId = await saveProgress(ctx, { - kind: 'relay-feature-guardian:progress', - version: 2, - checkedIds: [...checkedIds], - cycleStartedAt, - totalFeatures, - lastPost: { featureId: feature.id, ts }, + } catch (err) { + ctx.log('error', 'relay-feature-guardian.post-failed', { + channel, + feature: feature.id, + err: String(err), }); - if (!checkpointId) { - ctx.log('error', 'relay-feature-guardian.progress-checkpoint-failed', { - channel, - feature: feature.id, - ts, - }); - return; - } - ctx.log('info', 'relay-feature-guardian.posted', { + return; + } + const ts = deliveredSlackTs(result); + if (!ts) { + ctx.log('error', 'relay-feature-guardian.post-failed', { channel, feature: feature.id }); + return; + } + + // Checkpoint immediately after the confirmed provider receipt. The stable + // idempotency key makes a retry safe if this save times out or the run caps. + checkedIds.add(feature.id); + const completed: ProgressState = { + kind: 'relay-feature-guardian:progress', + version: 3, + generation: progress.state.generation, + checkedIds: [...checkedIds], + cycleStartedAt: progress.state.cycleStartedAt, + totalFeatures, + lastPost: { featureId: feature.id, ts }, + }; + let checkpoint: ProgressSnapshot; + try { + checkpoint = await store.save(completed, progress, features); + } catch (err) { + ctx.log('error', 'relay-feature-guardian.progress-checkpoint-failed', { channel, feature: feature.id, ts, - checkpointId, + err: String(err), }); - }, + return; + } + ctx.log('info', 'relay-feature-guardian.posted', { + channel, + feature: feature.id, + ts, + checkpointRevision: checkpoint.revision, + }); +} + +export default defineAgent({ + schedules: [{ name: 'hourly-check', cron: '0 * * * *', tz: 'America/New_York' }], + handler: runGuardian, }); diff --git a/CHANGELOG.md b/CHANGELOG.md index e6964f2d7..71f387c82 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,7 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `agent-relay integration` now discovers relayfile control-plane capabilities before sending API v3 headers, fails fast with upgrade and restart guidance for incompatible daemons, and safely replaces stale daemons when a compatible binary is installed. - `AgentRelaySDK` now maps Relaycast lifecycle states onto its existing Swift presence states, so root-package consumers compile with Relaycast 6.1 and later while package-local builds remain compatible with 6.0.5. -- `relay-feature-guardian` now reads the scoped Relay clone, posts to its configured channel, and checkpoints receipt-confirmed cycle progress so retries do not repeat a feature check. +- `relay-feature-guardian` now reads the scoped Relay clone, posts to its configured channel, and advances its exact, revision-safe cycle checkpoint only after a bounded wait returns a real Slack receipt, while safely reconciling retired manifest features. ## [10.6.3] - 2026-07-17