From 5e8bc0925c8af4c1f71045969f926706754bbd3b Mon Sep 17 00:00:00 2001 From: Joncallim <64296013+Joncallim@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:27:40 +0800 Subject: [PATCH 1/8] fix: preserve legacy clarification storage --- .../architect-legacy-clarification.test.ts | 390 ++++++++++++++++++ web/worker/orchestrator.ts | 28 +- 2 files changed, 411 insertions(+), 7 deletions(-) create mode 100644 web/__tests__/architect-legacy-clarification.test.ts diff --git a/web/__tests__/architect-legacy-clarification.test.ts b/web/__tests__/architect-legacy-clarification.test.ts new file mode 100644 index 00000000..0b45a939 --- /dev/null +++ b/web/__tests__/architect-legacy-clarification.test.ts @@ -0,0 +1,390 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const { + mockBuildWebResearchContext, + mockDbDelete, + mockDbInsert, + mockDbSelect, + mockDbUpdate, + mockGetModel, + mockGetProjectMcpOverview, + mockGetProvider, + mockGetWorkspaceSettings, + mockLoadCurrentProjectFilesystemDecision, + mockPublishTaskEvent, + mockReadS4RuntimeModeV1, + mockRecordArchitectPlanVersion, + mockRecordTaskLogBestEffort, + mockStreamText, + mockUpdateTaskStatusIfCurrent, + mockWriteArchitectCheckpointSafely, +} = vi.hoisted(() => ({ + mockBuildWebResearchContext: vi.fn(), + mockDbDelete: vi.fn(), + mockDbInsert: vi.fn(), + mockDbSelect: vi.fn(), + mockDbUpdate: vi.fn(), + mockGetModel: vi.fn(), + mockGetProjectMcpOverview: vi.fn(), + mockGetProvider: vi.fn(), + mockGetWorkspaceSettings: vi.fn(), + mockLoadCurrentProjectFilesystemDecision: vi.fn(), + mockPublishTaskEvent: vi.fn(), + mockReadS4RuntimeModeV1: vi.fn(), + mockRecordArchitectPlanVersion: vi.fn(), + mockRecordTaskLogBestEffort: vi.fn(), + mockStreamText: vi.fn(), + mockUpdateTaskStatusIfCurrent: vi.fn(), + mockWriteArchitectCheckpointSafely: vi.fn(), +})) + +vi.mock('ai', () => ({ streamText: mockStreamText })) + +vi.mock('@/db', () => ({ + db: { + delete: mockDbDelete, + insert: mockDbInsert, + select: mockDbSelect, + update: mockDbUpdate, + }, +})) + +vi.mock('@/lib/providers/registry', () => ({ + getModel: mockGetModel, + getProvider: mockGetProvider, + providerExecutionSnapshot: (config: { + id: string + isLocal: boolean + modelId: string + providerType: string + updatedAt: Date + }) => ({ + acpExecutionMode: 'not_applicable', + configId: config.id, + fingerprint: 'provider-snapshot', + isLocal: config.isLocal, + modelId: config.modelId, + providerType: config.providerType, + updatedAt: config.updatedAt, + }), +})) + +vi.mock('@/lib/providers/default', () => ({ + resolveDefaultProvider: vi.fn(), +})) + +vi.mock('@/lib/mcps/manager', () => ({ + getProjectMcpOverview: mockGetProjectMcpOverview, +})) + +vi.mock('@/lib/mcps/filesystem-grant-reconciliation', () => ({ + loadCurrentProjectFilesystemDecision: mockLoadCurrentProjectFilesystemDecision, +})) + +vi.mock('@/lib/workspace', () => ({ + displayPathForWorkspacePath: vi.fn(), + getWorkspaceSettings: mockGetWorkspaceSettings, +})) + +vi.mock('@/worker/architect-context', () => ({ + buildSpecialistContext: vi.fn(() => ''), + buildWebResearchContext: mockBuildWebResearchContext, + detectSoftwareProfile: vi.fn(() => ({ kind: 'software' })), +})) + +vi.mock('@/worker/events', () => ({ + publishTaskEvent: mockPublishTaskEvent, +})) + +vi.mock('@/worker/task-logs', () => ({ + recordTaskLogBestEffort: mockRecordTaskLogBestEffort, +})) + +vi.mock('@/worker/task-state', () => ({ + updateTaskStatus: vi.fn(), + updateTaskStatusIfCurrent: mockUpdateTaskStatusIfCurrent, +})) + +vi.mock('@/worker/checkpoints', () => ({ + readLatestArchitectCheckpointSafely: vi.fn().mockResolvedValue(null), + writeArchitectCheckpointSafely: mockWriteArchitectCheckpointSafely, +})) + +vi.mock('@/worker/workforce-materializer', () => ({ + materializeWorkforceFromArchitectArtifact: vi.fn(), +})) + +vi.mock('@/lib/mcps/s4-lease', () => ({ + readS4RuntimeModeV1: mockReadS4RuntimeModeV1, +})) + +vi.mock('@/lib/mcps/s4-protocol-store', async (importOriginal) => ({ + ...await importOriginal(), + bindArchitectReplanContext: vi.fn(), + recordArchitectPlanVersion: mockRecordArchitectPlanVersion, + resolveArchitectReplanEntry: vi.fn(), +})) + +type ChainHooks = { + set?: (value: unknown) => void + values?: (value: unknown) => void +} + +function chain(resolveValue: unknown, hooks: ChainHooks = {}) { + const thenable: Record = { + then: (onFulfilled: (value: unknown) => unknown, onRejected?: (error: unknown) => unknown) => + Promise.resolve(resolveValue).then(onFulfilled, onRejected), + } + for (const method of [ + 'from', + 'innerJoin', + 'limit', + 'orderBy', + 'returning', + 'set', + 'values', + 'where', + ]) { + thenable[method] = (value: unknown) => { + if (method === 'set') hooks.set?.(value) + if (method === 'values') hooks.values?.(value) + return thenable + } + } + return thenable +} + +const taskId = '11111111-1111-4111-8111-111111111111' +const projectId = '22222222-2222-4222-8222-222222222222' +const runId = '33333333-3333-4333-8333-333333333333' +const artifactId = '44444444-4444-4444-8444-444444444444' +const questionId = '55555555-5555-4555-8555-555555555555' +const questionText = 'Which branch should receive the change?' +const planText = [ + '# Implementation plan', + '', + 'Confirm the target branch before changing the repository.', + '', + '```open_questions_json', + JSON.stringify({ + questions: [{ question: questionText, suggestions: ['main', 'release'] }], + }), + '```', +].join('\n') + +describe('Architect clarification storage mode', () => { + const priorEnv = { + digestKey: process.env.FORGE_ARCHITECT_PLAN_DIGEST_KEY_HEX, + digestKeyId: process.env.FORGE_ARCHITECT_PLAN_DIGEST_KEY_ID, + mockArchitect: process.env.FORGE_WORKER_MOCK_ARCHITECT, + writerUrl: process.env.FORGE_ARCHITECT_PLAN_WRITER_DATABASE_URL, + } + + const selectResults: unknown[] = [] + const insertResults: unknown[] = [] + const updateResults: unknown[] = [] + const insertedValues: unknown[] = [] + + beforeEach(() => { + vi.clearAllMocks() + vi.resetModules() + selectResults.length = 0 + insertResults.length = 0 + updateResults.length = 0 + insertedValues.length = 0 + + delete process.env.FORGE_WORKER_MOCK_ARCHITECT + process.env.FORGE_ARCHITECT_PLAN_DIGEST_KEY_HEX = 'a'.repeat(64) + process.env.FORGE_ARCHITECT_PLAN_DIGEST_KEY_ID = 'test-key-v1' + process.env.FORGE_ARCHITECT_PLAN_WRITER_DATABASE_URL = 'postgresql://writer/test' + + mockDbSelect.mockImplementation(() => chain(selectResults.shift() ?? [])) + mockDbInsert.mockImplementation(() => chain(insertResults.shift() ?? [], { + values: (value) => insertedValues.push(value), + })) + mockDbUpdate.mockImplementation(() => chain(updateResults.shift() ?? [])) + mockDbDelete.mockImplementation(() => chain(undefined)) + mockGetModel.mockResolvedValue({}) + mockGetProvider.mockResolvedValue({ + config: { + displayName: 'Test provider', + id: 'provider-1', + isLocal: false, + modelId: 'test-model', + providerType: 'openai', + updatedAt: new Date('2026-07-30T00:00:00.000Z'), + }, + }) + mockGetProjectMcpOverview.mockResolvedValue({ + catalog: [], + missingRequired: [], + statuses: [], + warnings: [], + }) + mockGetWorkspaceSettings.mockResolvedValue({}) + mockLoadCurrentProjectFilesystemDecision.mockResolvedValue(null) + mockBuildWebResearchContext.mockResolvedValue('') + mockPublishTaskEvent.mockResolvedValue(undefined) + mockRecordTaskLogBestEffort.mockResolvedValue(undefined) + mockUpdateTaskStatusIfCurrent.mockResolvedValue(true) + mockWriteArchitectCheckpointSafely.mockResolvedValue({ + latestPath: '/tmp/latest.md', + runPath: '/tmp/run.md', + }) + mockStreamText.mockReturnValue({ + finishReason: Promise.resolve('stop'), + textStream: (async function* () { + yield planText + })(), + usage: Promise.resolve({ inputTokens: 10, outputTokens: 20 }), + }) + }) + + afterEach(() => { + for (const [name, value] of [ + ['FORGE_ARCHITECT_PLAN_DIGEST_KEY_HEX', priorEnv.digestKey], + ['FORGE_ARCHITECT_PLAN_DIGEST_KEY_ID', priorEnv.digestKeyId], + ['FORGE_WORKER_MOCK_ARCHITECT', priorEnv.mockArchitect], + ['FORGE_ARCHITECT_PLAN_WRITER_DATABASE_URL', priorEnv.writerUrl], + ] as const) { + if (value === undefined) delete process.env[name] + else process.env[name] = value + } + }) + + async function runOpenQuestionFlow(mode: 'legacy' | 'protected') { + const task = { + id: taskId, + pmProviderConfigId: 'provider-1', + projectId, + prompt: 'Plan this change.', + status: 'pending', + title: 'Clarification storage', + } + const project = { + defaultBranch: 'main', + githubRepo: 'owner/repo', + id: projectId, + localPath: null, + name: 'Clarification project', + } + const run = { + agentType: 'architect', + id: runId, + modelIdUsed: 'test-model', + providerConfigId: 'provider-1', + startedAt: new Date('2026-07-30T00:00:00.000Z'), + status: 'running', + taskId, + } + const artifact = { + agentRunId: runId, + artifactType: 'adr_text', + content: mode === 'protected' + ? 'Architect plan available in protected history' + : 'Confirm the target branch before changing the repository.', + createdAt: new Date('2026-07-30T00:01:00.000Z'), + id: artifactId, + metadata: mode === 'protected' + ? { historyAvailable: true, planVersion: '1' } + : { historyAvailable: false, planVersion: '1', storageMode: 'legacy' }, + } + const question = { + id: questionId, + status: mode === 'protected' ? 'open' : 'legacy_unavailable', + taskId, + } + + selectResults.push( + [{ project, task }], + [{ + agentType: 'architect', + displayName: 'Architect', + id: 'architect-config', + isActive: true, + providerConfigId: 'provider-1', + systemPrompt: 'Plan carefully.', + }], + [], + [], + [], + ) + if (mode === 'legacy') { + insertResults.push([run], [artifact], [question]) + updateResults.push([{ id: runId }]) + } else { + insertResults.push([run], [question]) + updateResults.push([artifact], [{ id: runId }]) + } + + mockReadS4RuntimeModeV1.mockResolvedValue(mode) + mockRecordArchitectPlanVersion.mockResolvedValue({ + artifactId, + entries: [{ + agent: null, + bindingFingerprint: null, + content: 'Confirm the target branch before changing the repository.', + contentDigest: `hmac-sha256:${'b'.repeat(64)}`, + digestKeyId: 'test-key-v1', + entryId: 'plan_body:000000', + entryKind: 'plan_body', + planArtifactId: artifactId, + planVersion: '1', + projectionEligible: false, + requirementKey: null, + schemaVersion: 1, + taskId, + }], + entrySetDigest: `hmac-sha256:${'c'.repeat(64)}`, + }) + + const orchestrator = await import('@/worker/orchestrator') + await expect(orchestrator.processTask(taskId)).resolves.toBe('completed') + return { artifact, orchestrator } + } + + it('keeps legacy open questions on the legacy artifact path and advances to awaiting_answers', async () => { + const { artifact, orchestrator } = await runOpenQuestionFlow('legacy') + const questionValues = insertedValues.find(Array.isArray) as Array> + + expect(mockRecordArchitectPlanVersion).not.toHaveBeenCalled() + expect(questionValues).toEqual([ + expect.objectContaining({ + id: expect.any(String), + status: 'legacy_unavailable', + taskId, + }), + ]) + expect(questionValues[0]).not.toHaveProperty('questionEntryId') + expect(questionValues[0]).not.toHaveProperty('sourcePlanArtifactId') + expect(questionValues[0]).not.toHaveProperty('sourcePlanVersion') + expect(mockUpdateTaskStatusIfCurrent).toHaveBeenNthCalledWith(1, taskId, 'pending', 'running') + expect(mockUpdateTaskStatusIfCurrent).toHaveBeenNthCalledWith(2, taskId, 'running', 'awaiting_answers') + await expect(orchestrator.previousPlanForArchitectRun({ + agentRunId: runId, + artifact, + checkpoint: null, + taskId, + })).resolves.toBe(artifact.content) + }) + + it('retains protected clarification bindings for protected artifacts', async () => { + await runOpenQuestionFlow('protected') + const questionValues = insertedValues.find(Array.isArray) as Array> + const persistedQuestionId = questionValues[0]?.id + + expect(mockRecordArchitectPlanVersion).toHaveBeenCalledOnce() + expect(persistedQuestionId).toEqual(expect.any(String)) + expect(questionValues).toEqual([ + expect.objectContaining({ + id: persistedQuestionId, + questionEntryId: `clarification_question:${persistedQuestionId}`, + sourcePlanArtifactId: artifactId, + sourcePlanVersion: 1, + status: 'open', + taskId, + }), + ]) + expect(mockUpdateTaskStatusIfCurrent).toHaveBeenNthCalledWith(2, taskId, 'running', 'awaiting_answers') + }) +}) diff --git a/web/worker/orchestrator.ts b/web/worker/orchestrator.ts index 64a6c853..6585f399 100644 --- a/web/worker/orchestrator.ts +++ b/web/worker/orchestrator.ts @@ -544,6 +544,7 @@ async function loadLatestPlanArtifact(taskId: string): Promise { // Answered rows are the opaque durable projection of protected subledger @@ -807,12 +812,20 @@ async function persistOpenQuestions( const rows = await db .insert(taskQuestions) .values( - questions.map((question) => ({ - id: question.questionId, taskId, - questionEntryId: `clarification_question:${question.questionId}`, - sourcePlanArtifactId: artifactId, sourcePlanVersion: Number(planVersion), - status: 'open' as const, - })), + questions.map((question) => storageMode === 'protected' + ? { + id: question.questionId, + taskId, + questionEntryId: `clarification_question:${question.questionId}`, + sourcePlanArtifactId: artifactId, + sourcePlanVersion: Number(planVersion), + status: 'open' as const, + } + : { + id: question.questionId, + taskId, + status: 'legacy_unavailable' as const, + }), ) .returning() @@ -1205,6 +1218,7 @@ async function runArchitect( protectedOpenQuestions, artifact.id, planVersion, + artifact.architectPlanStorageMode, claimLeaseFence, ) From 362e41939e3028d891e9d75898b9333c48b902ab Mon Sep 17 00:00:00 2001 From: Joncallim <64296013+Joncallim@users.noreply.github.com> Date: Thu, 30 Jul 2026 14:19:24 +0800 Subject: [PATCH 2/8] fix: complete legacy clarification lifecycle --- web/__tests__/api.test.ts | 201 +++++++++++- .../architect-legacy-clarification.test.ts | 206 +++++++++++- web/__tests__/task-page-retry-handoff.test.ts | 17 + web/app/api/tasks/[id]/questions/route.ts | 295 +++++++++++++++--- web/app/dashboard/tasks/[id]/page.tsx | 9 +- .../0027_epic_172_s4_packet_context.sql | 5 +- web/db/schema.ts | 2 +- web/lib/mcps/legacy-clarification.ts | 152 +++++++++ web/worker/orchestrator.ts | 89 +++++- 9 files changed, 930 insertions(+), 46 deletions(-) create mode 100644 web/lib/mcps/legacy-clarification.ts diff --git a/web/__tests__/api.test.ts b/web/__tests__/api.test.ts index c043e9c7..02f5d1d3 100644 --- a/web/__tests__/api.test.ts +++ b/web/__tests__/api.test.ts @@ -7188,7 +7188,13 @@ describe('PUT /api/tasks/:id/filesystem-grants — explicit grant approvals', () // --------------------------------------------------------------------------- describe('POST /api/tasks/:id/questions', () => { - beforeEach(() => { vi.clearAllMocks() }) + beforeEach(() => { + vi.clearAllMocks() + mockReadS4RuntimeModeV1.mockResolvedValue('protected') + mockArchitectPlanStorageConfiguration.mockReturnValue({ + mode: 'protected', digestKey: Buffer.alloc(32, 7), digestKeyId: 'test-v1', + }) + }) it('returns 401 when not authenticated', async () => { mockGetSession.mockResolvedValue(null) @@ -7245,6 +7251,66 @@ describe('POST /api/tasks/:id/questions', () => { expect(JSON.stringify(body)).not.toContain('RAW-') }) + it('presents an encrypted legacy clarification only through the authorized question route', async () => { + const previousSecret = process.env.SESSION_SECRET + process.env.SESSION_SECRET = 'legacy-question-route-test-secret' + try { + const taskId = '11111111-1111-4111-8111-111111111111' + const questionId = '77777777-7777-4777-8777-777777777777' + const agentRunId = '88888888-8888-4888-8888-888888888888' + const question = 'RAW-LEGACY-QUESTION-SENTINEL' + const suggestion = 'RAW-LEGACY-SUGGESTION-SENTINEL' + const { sealLegacyClarification } = await import('@/lib/mcps/legacy-clarification') + const metadata = { + historyAvailable: false, + planVersion: '1', + storageMode: 'legacy', + legacyClarificationV1: sealLegacyClarification({ + schemaVersion: 1, + taskId, + agentRunId, + planVersion: '1', + questions: [{ id: questionId, question, suggestions: [suggestion], answer: null }], + }), + } + mockGetSession.mockResolvedValue(FAKE_SESSION) + mockReadS4RuntimeModeV1.mockResolvedValueOnce('legacy') + mockArchitectPlanStorageConfiguration.mockReturnValueOnce({ mode: 'legacy' }) + mockDbSelect + .mockReturnValueOnce(chain([{ id: taskId, status: 'awaiting_answers' }])) + .mockReturnValueOnce(chain([{ + id: questionId, + status: 'open', + createdAt: new Date('2026-07-30T00:00:00.000Z'), + answeredAt: null, + }])) + .mockReturnValueOnce(chain([{ id: 'artifact-1', agentRunId, metadata }])) + + const { GET } = await import('@/app/api/tasks/[id]/questions/route') + const response = await GET(authRequest(`/api/tasks/${taskId}/questions`) as never, { + params: Promise.resolve({ id: taskId }), + }) + + expect(response.status).toBe(200) + expect(await response.json()).toEqual({ + questions: [{ + id: questionId, + status: 'open', + createdAt: '2026-07-30T00:00:00.000Z', + answeredAt: null, + question, + suggestions: [suggestion], + answer: null, + }], + }) + expect(JSON.stringify(metadata)).not.toContain(question) + expect(JSON.stringify(metadata)).not.toContain(suggestion) + } finally { + if (previousSecret === undefined) delete process.env.SESSION_SECRET + else process.env.SESSION_SECRET = previousSecret + } + }) + it('accepts an opaque question id and answer but returns only content-free summaries', async () => { mockGetSession.mockResolvedValue(FAKE_SESSION) const questionId = '77777777-7777-4777-8777-777777777777' @@ -7294,6 +7360,139 @@ describe('POST /api/tasks/:id/questions', () => { expect(JSON.parse(answeredEvent?.[5] as string)).toEqual({ answeredCount: 1, allAnswered: true }) expect(JSON.stringify(mockRedisEval.mock.calls)).not.toContain('RAW-') }) + + it('records a legacy answer in the encrypted artifact and queues re-plan without public plaintext', async () => { + const previousSecret = process.env.SESSION_SECRET + process.env.SESSION_SECRET = 'legacy-question-answer-test-secret' + try { + const taskId = '11111111-1111-4111-8111-111111111111' + const questionId = '77777777-7777-4777-8777-777777777777' + const agentRunId = '88888888-8888-4888-8888-888888888888' + const question = 'RAW-LEGACY-QUESTION-SENTINEL' + const answer = 'RAW-LEGACY-ANSWER-SENTINEL' + const { readLegacyClarification, sealLegacyClarification } = await import('@/lib/mcps/legacy-clarification') + const metadata = { + historyAvailable: false, + planVersion: '1', + storageMode: 'legacy', + legacyClarificationV1: sealLegacyClarification({ + schemaVersion: 1, + taskId, + agentRunId, + planVersion: '1', + questions: [{ id: questionId, question, suggestions: ['main'], answer: null }], + }), + } + const artifactUpdate = chain([{ id: 'artifact-1' }]) + let artifactSet: Record = {} + artifactUpdate.set = vi.fn((value: Record) => { + artifactSet = value + return artifactUpdate + }) + const questionUpdate = chain([{ + id: questionId, + status: 'answered', + createdAt: new Date('2026-07-30T00:00:00.000Z'), + answeredAt: new Date('2026-07-30T00:01:00.000Z'), + }]) + let questionSet: Record = {} + questionUpdate.set = vi.fn((value: Record) => { + questionSet = value + return questionUpdate + }) + + mockGetSession.mockResolvedValue(FAKE_SESSION) + mockReadS4RuntimeModeV1.mockResolvedValueOnce('legacy') + mockArchitectPlanStorageConfiguration.mockReturnValueOnce({ mode: 'legacy' }) + mockDbSelect + .mockReturnValueOnce(chain([{ id: taskId, status: 'awaiting_answers' }])) + .mockReturnValueOnce(chain([{ + id: questionId, + status: 'open', + createdAt: new Date('2026-07-30T00:00:00.000Z'), + answeredAt: null, + answerReferenceId: null, + questionEntryId: null, + sourcePlanArtifactId: null, + sourcePlanVersion: null, + }])) + .mockReturnValueOnce(chain([{ id: 'artifact-1', agentRunId, metadata }])) + mockDbUpdate + .mockReturnValueOnce(artifactUpdate) + .mockReturnValueOnce(questionUpdate) + mockRedisLpush.mockResolvedValue(1) + mockRedisEval.mockResolvedValue(1) + + const { POST } = await import('@/app/api/tasks/[id]/questions/route') + const response = await POST(authRequest(`/api/tasks/${taskId}/questions`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ answers: [{ id: questionId, answer }] }), + }) as never, { params: Promise.resolve({ id: taskId }) }) + + expect(response.status).toBe(200) + expect(await response.json()).toEqual({ + questions: [{ + id: questionId, + status: 'answered', + createdAt: '2026-07-30T00:00:00.000Z', + answeredAt: '2026-07-30T00:01:00.000Z', + }], + allAnswered: true, + }) + expect(questionSet).toEqual(expect.objectContaining({ + status: 'answered', + answeredBy: FAKE_SESSION.userId, + answeredAt: expect.any(Date), + })) + const storedMetadata = artifactSet.metadata as Record + expect(JSON.stringify(storedMetadata)).not.toContain(question) + expect(JSON.stringify(storedMetadata)).not.toContain(answer) + const persisted = readLegacyClarification(storedMetadata, { + taskId, + agentRunId, + planVersion: '1', + }) + expect(persisted?.questions).toEqual([{ + id: questionId, + question, + suggestions: ['main'], + answer, + }]) + expect(mockAppendArchitectClarificationAnswer).not.toHaveBeenCalled() + expect(mockRedisLpush).toHaveBeenCalledWith('forge:answers', JSON.stringify({ taskId })) + expect(JSON.stringify(mockRedisLpush.mock.calls)).not.toContain(answer) + expect(JSON.stringify(mockRedisEval.mock.calls)).not.toContain(answer) + } finally { + if (previousSecret === undefined) delete process.env.SESSION_SECRET + else process.env.SESSION_SECRET = previousSecret + } + }) + + it('does not fall back to legacy answering for a protected clarification', async () => { + mockGetSession.mockResolvedValue(FAKE_SESSION) + const taskId = '11111111-1111-4111-8111-111111111111' + const questionId = '77777777-7777-4777-8777-777777777777' + mockDbSelect + .mockReturnValueOnce(chain([{ id: taskId, status: 'awaiting_answers' }])) + .mockReturnValueOnce(chain([{ + id: questionId, + sourcePlanArtifactId: null, + sourcePlanVersion: null, + }])) + + const { POST } = await import('@/app/api/tasks/[id]/questions/route') + const response = await POST(authRequest(`/api/tasks/${taskId}/questions`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ answers: [{ id: questionId, answer: 'protected answer' }] }), + }) as never, { params: Promise.resolve({ id: taskId }) }) + + expect(response.status).toBe(409) + expect(mockDbTransaction).not.toHaveBeenCalled() + expect(mockDbUpdate).not.toHaveBeenCalled() + expect(mockAppendArchitectClarificationAnswer).not.toHaveBeenCalled() + }) }) // --------------------------------------------------------------------------- diff --git a/web/__tests__/architect-legacy-clarification.test.ts b/web/__tests__/architect-legacy-clarification.test.ts index 0b45a939..a47cd863 100644 --- a/web/__tests__/architect-legacy-clarification.test.ts +++ b/web/__tests__/architect-legacy-clarification.test.ts @@ -1,4 +1,6 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import fs from 'node:fs' +import path from 'node:path' const { mockBuildWebResearchContext, @@ -177,6 +179,7 @@ describe('Architect clarification storage mode', () => { digestKey: process.env.FORGE_ARCHITECT_PLAN_DIGEST_KEY_HEX, digestKeyId: process.env.FORGE_ARCHITECT_PLAN_DIGEST_KEY_ID, mockArchitect: process.env.FORGE_WORKER_MOCK_ARCHITECT, + sessionSecret: process.env.SESSION_SECRET, writerUrl: process.env.FORGE_ARCHITECT_PLAN_WRITER_DATABASE_URL, } @@ -197,6 +200,7 @@ describe('Architect clarification storage mode', () => { process.env.FORGE_ARCHITECT_PLAN_DIGEST_KEY_HEX = 'a'.repeat(64) process.env.FORGE_ARCHITECT_PLAN_DIGEST_KEY_ID = 'test-key-v1' process.env.FORGE_ARCHITECT_PLAN_WRITER_DATABASE_URL = 'postgresql://writer/test' + process.env.SESSION_SECRET = 'legacy-clarification-test-secret' mockDbSelect.mockImplementation(() => chain(selectResults.shift() ?? [])) mockDbInsert.mockImplementation(() => chain(insertResults.shift() ?? [], { @@ -245,6 +249,7 @@ describe('Architect clarification storage mode', () => { ['FORGE_ARCHITECT_PLAN_DIGEST_KEY_HEX', priorEnv.digestKey], ['FORGE_ARCHITECT_PLAN_DIGEST_KEY_ID', priorEnv.digestKeyId], ['FORGE_WORKER_MOCK_ARCHITECT', priorEnv.mockArchitect], + ['SESSION_SECRET', priorEnv.sessionSecret], ['FORGE_ARCHITECT_PLAN_WRITER_DATABASE_URL', priorEnv.writerUrl], ] as const) { if (value === undefined) delete process.env[name] @@ -291,7 +296,7 @@ describe('Architect clarification storage mode', () => { } const question = { id: questionId, - status: mode === 'protected' ? 'open' : 'legacy_unavailable', + status: 'open', taskId, } @@ -351,13 +356,26 @@ describe('Architect clarification storage mode', () => { expect(questionValues).toEqual([ expect.objectContaining({ id: expect.any(String), - status: 'legacy_unavailable', + status: 'open', taskId, }), ]) expect(questionValues[0]).not.toHaveProperty('questionEntryId') expect(questionValues[0]).not.toHaveProperty('sourcePlanArtifactId') expect(questionValues[0]).not.toHaveProperty('sourcePlanVersion') + const legacyArtifactInsert = insertedValues.find((value) => + value && !Array.isArray(value) && (value as { artifactType?: unknown }).artifactType === 'adr_text', + ) as { metadata: Record } + expect(JSON.stringify(legacyArtifactInsert.metadata)).not.toContain(questionText) + expect(legacyArtifactInsert.metadata.legacyClarificationV1).toEqual({ + schemaVersion: 1, + ciphertext: expect.stringMatching(/^v1:/), + }) + expect(mockWriteArchitectCheckpointSafely).toHaveBeenCalledWith(expect.objectContaining({ + openQuestionCount: 1, + openQuestions: [], + })) + expect(JSON.stringify(mockWriteArchitectCheckpointSafely.mock.calls)).not.toContain(questionText) expect(mockUpdateTaskStatusIfCurrent).toHaveBeenNthCalledWith(1, taskId, 'pending', 'running') expect(mockUpdateTaskStatusIfCurrent).toHaveBeenNthCalledWith(2, taskId, 'running', 'awaiting_answers') await expect(orchestrator.previousPlanForArchitectRun({ @@ -368,6 +386,21 @@ describe('Architect clarification storage mode', () => { })).resolves.toBe(artifact.content) }) + it('keeps public plaintext null while permitting open-to-answered legacy projection state', () => { + const migration = fs.readFileSync( + path.join(process.cwd(), 'db/migrations/0027_epic_172_s4_packet_context.sql'), + 'utf8', + ) + const constraint = migration.slice( + migration.indexOf('ADD CONSTRAINT task_questions_no_public_plaintext_chk'), + migration.indexOf('CREATE TABLE public.architect_clarification_answer_writes'), + ) + expect(constraint).toContain('question IS NULL AND suggestions IS NULL AND answer IS NULL') + expect(constraint).toContain("status IN ('legacy_unavailable', 'open') AND answered_at IS NULL") + expect(constraint).toContain("status = 'answered' AND answered_at IS NOT NULL") + expect(constraint).toContain('answer_reference_id IS NULL') + }) + it('retains protected clarification bindings for protected artifacts', async () => { await runOpenQuestionFlow('protected') const questionValues = insertedValues.find(Array.isArray) as Array> @@ -387,4 +420,173 @@ describe('Architect clarification storage mode', () => { ]) expect(mockUpdateTaskStatusIfCurrent).toHaveBeenNthCalledWith(2, taskId, 'running', 'awaiting_answers') }) + + it('replans from the answered encrypted legacy artifact without copying plaintext to public sinks', async () => { + const answerText = 'Use the release branch.' + const { sealLegacyClarification } = await import('@/lib/mcps/legacy-clarification') + const routingMetadata = { + agentBreakdown: [{ + role: 'Backend', + tasks: 1, + summary: 'Apply the targeted branch change', + }], + agentBreakdownSource: 'fence', + capabilityClassification: { + schemaVersion: 1, + required: ['business-logic'], + optional: [], + excluded: [], + }, + mcpExecutionDesign: { + proposed: { + schemaVersion: 1, + requirements: [], + promptOverlays: {}, + requirementContexts: [], + mcpAwareSubtasks: [], + }, + }, + } + const priorArtifact = { + agentRunId: runId, + content: 'Confirm the target branch before changing the repository.', + id: artifactId, + metadata: { + ...routingMetadata, + historyAvailable: false, + planVersion: '1', + storageMode: 'legacy', + legacyClarificationV1: sealLegacyClarification({ + schemaVersion: 1, + taskId, + agentRunId: runId, + planVersion: '1', + questions: [{ + id: questionId, + question: questionText, + suggestions: ['main', 'release'], + answer: answerText, + }], + }), + }, + } + const task = { + id: taskId, + pmProviderConfigId: 'provider-1', + projectId, + prompt: 'Plan this change.', + status: 'awaiting_answers', + title: 'Clarification storage', + } + const project = { + defaultBranch: 'main', + githubRepo: 'owner/repo', + id: projectId, + localPath: null, + name: 'Clarification project', + } + const revisedPlan = [ + '# Implementation plan', + '', + 'Confirm the target branch before changing the repository.', + '', + '```agent_breakdown_json', + JSON.stringify({ agents: routingMetadata.agentBreakdown }), + '```', + '', + '```capability_classification_json', + JSON.stringify(routingMetadata.capabilityClassification), + '```', + '', + '```mcp_execution_design_json', + JSON.stringify(routingMetadata.mcpExecutionDesign.proposed), + '```', + '', + '```open_questions_json', + '{"questions":[]}', + '```', + ].join('\n') + mockStreamText.mockImplementation((input: { prompt: string }) => { + expect(input.prompt).toContain(questionText) + expect(input.prompt).toContain(answerText) + return { + finishReason: Promise.resolve('stop'), + textStream: (async function* () { + yield revisedPlan + })(), + usage: Promise.resolve({ inputTokens: 12, outputTokens: 24 }), + } + }) + selectResults.push( + [{ project, task }], + [{ + id: questionId, + taskId, + status: 'answered', + createdAt: new Date('2026-07-30T00:00:00.000Z'), + answeredAt: new Date('2026-07-30T00:01:00.000Z'), + answeredBy: '66666666-6666-4666-8666-666666666666', + answerReferenceId: null, + questionEntryId: null, + sourcePlanArtifactId: null, + sourcePlanVersion: null, + }], + [priorArtifact], + [{ + agentType: 'architect', + displayName: 'Architect', + id: 'architect-config', + isActive: true, + providerConfigId: 'provider-1', + systemPrompt: 'Plan carefully.', + }], + [priorArtifact], + [], + [], + ) + insertResults.push( + [{ + agentType: 'architect', + id: '77777777-7777-4777-8777-777777777777', + modelIdUsed: 'test-model', + providerConfigId: 'provider-1', + startedAt: new Date('2026-07-30T00:02:00.000Z'), + status: 'running', + taskId, + }], + [{ + agentRunId: '77777777-7777-4777-8777-777777777777', + artifactType: 'adr_text', + content: 'Confirm the target branch before changing the repository.', + createdAt: new Date('2026-07-30T00:03:00.000Z'), + id: '88888888-8888-4888-8888-888888888888', + metadata: { + ...routingMetadata, + historyAvailable: false, + planVersion: '2', + storageMode: 'legacy', + }, + }], + ) + updateResults.push([{ id: '77777777-7777-4777-8777-777777777777' }]) + mockReadS4RuntimeModeV1.mockResolvedValue('legacy') + + const { processAnsweredQuestions } = await import('@/worker/orchestrator') + await expect(processAnsweredQuestions(taskId)).resolves.toBe('completed') + + expect(mockUpdateTaskStatusIfCurrent).toHaveBeenNthCalledWith( + 1, taskId, 'awaiting_answers', 'running', + ) + expect(mockUpdateTaskStatusIfCurrent).toHaveBeenNthCalledWith( + 2, taskId, 'running', 'awaiting_approval', + ) + expect(JSON.stringify(insertedValues)).not.toContain(questionText) + expect(JSON.stringify(insertedValues)).not.toContain(answerText) + expect(JSON.stringify(mockPublishTaskEvent.mock.calls)).not.toContain(questionText) + expect(JSON.stringify(mockPublishTaskEvent.mock.calls)).not.toContain(answerText) + expect(JSON.stringify(mockRecordTaskLogBestEffort.mock.calls)).not.toContain(questionText) + expect(JSON.stringify(mockRecordTaskLogBestEffort.mock.calls)).not.toContain(answerText) + expect(JSON.stringify(mockWriteArchitectCheckpointSafely.mock.calls)).not.toContain(questionText) + expect(JSON.stringify(mockWriteArchitectCheckpointSafely.mock.calls)).not.toContain(answerText) + }) }) diff --git a/web/__tests__/task-page-retry-handoff.test.ts b/web/__tests__/task-page-retry-handoff.test.ts index c75e6098..bf2b4948 100644 --- a/web/__tests__/task-page-retry-handoff.test.ts +++ b/web/__tests__/task-page-retry-handoff.test.ts @@ -1,4 +1,6 @@ import { describe, expect, it, vi } from 'vitest' +import fs from 'node:fs' +import path from 'node:path' vi.mock('next/navigation', () => ({ useParams: vi.fn(), @@ -32,6 +34,21 @@ import { } from '@/app/dashboard/tasks/[id]/page' describe('task page retry handoff controls', () => { + it('loads authorized legacy clarification text when protected history has no plan version', () => { + const source = fs.readFileSync( + path.join(process.cwd(), 'app/dashboard/tasks/[id]/page.tsx'), + 'utf8', + ) + const loader = source.slice( + source.indexOf('const loadClarificationHistory'), + source.indexOf('const loadTask = useCallback'), + ) + expect(loader).toContain('if (!planVersion)') + expect(loader).toContain('fetch(`/api/tasks/${taskId}/questions`)') + expect(loader).toContain('setClarificationQuestions(body.questions ?? [])') + expect(loader).toContain('fetch(`/api/tasks/${taskId}/architect-plan-history/${planVersion}`)') + }) + it('omits a pointer for D1 and preserves the exact D1 tuple for D2 reapproval', () => { expect(filesystemGrantExpectedPointerFromState({ currentDecision: null, diff --git a/web/app/api/tasks/[id]/questions/route.ts b/web/app/api/tasks/[id]/questions/route.ts index c24f2d40..56a0c8ef 100644 --- a/web/app/api/tasks/[id]/questions/route.ts +++ b/web/app/api/tasks/[id]/questions/route.ts @@ -2,8 +2,8 @@ import { NextResponse } from 'next/server' import type { NextRequest } from 'next/server' import { z } from 'zod' import { db } from '@/db' -import { taskQuestions } from '@/db/schema' -import { and, asc, eq, inArray } from 'drizzle-orm' +import { agentRuns, artifacts, taskQuestions } from '@/db/schema' +import { and, asc, desc, eq, inArray } from 'drizzle-orm' import { getSession, readSessionCredential } from '@/lib/session' import { redis } from '@/lib/redis' import { getAccessibleTask } from '@/lib/task-access' @@ -13,6 +13,14 @@ import { taskQuestionSummary } from '@/lib/mcps/clarification-projection' import { appendArchitectClarificationAnswer } from '@/lib/mcps/history-reader' import { architectPlanStorageConfiguration } from '@/lib/mcps/s4-protocol-store' import { readS4RuntimeModeV1 } from '@/lib/mcps/s4-lease' +import { + LEGACY_CLARIFICATION_MAX_TEXT_BYTES, + LEGACY_CLARIFICATION_METADATA_KEY, + answerLegacyClarification, + legacyClarificationAllAnswered, + readLegacyClarification, + sealLegacyClarification, +} from '@/lib/mcps/legacy-clarification' // --------------------------------------------------------------------------- // Validation schema @@ -23,12 +31,33 @@ const answersSchema = z.object({ .array( z.object({ id: z.string().uuid(), - answer: z.string().min(1, 'Answer cannot be empty'), + answer: z.string() + .min(1, 'Answer cannot be empty') + .refine( + (value) => Buffer.byteLength(value, 'utf8') <= LEGACY_CLARIFICATION_MAX_TEXT_BYTES, + 'Answer is too large', + ), }), ) - .min(1, 'At least one answer is required'), + .min(1, 'At least one answer is required') + .superRefine((answers, context) => { + const ids = new Set(answers.map((answer) => answer.id)) + if (ids.size !== answers.length) { + context.addIssue({ code: 'custom', message: 'Question ids must be unique' }) + } + }), }) +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +const legacyArtifactSelection = { + id: artifacts.id, + agentRunId: artifacts.agentRunId, + metadata: artifacts.metadata, +} + // --------------------------------------------------------------------------- // GET /api/tasks/:id/questions // --------------------------------------------------------------------------- @@ -62,7 +91,49 @@ export async function GET( .where(eq(taskQuestions.taskId, taskId)) .orderBy(asc(taskQuestions.createdAt)) - return NextResponse.json({ questions: questions.map(taskQuestionSummary) }) + const summaries = questions.map(taskQuestionSummary) + if (questions.length === 0) return NextResponse.json({ questions: summaries }) + const storage = architectPlanStorageConfiguration(process.env, await readS4RuntimeModeV1()) + if (storage.mode !== 'legacy') { + return NextResponse.json({ questions: summaries }) + } + + const [artifact] = await db + .select(legacyArtifactSelection) + .from(artifacts) + .innerJoin(agentRuns, eq(artifacts.agentRunId, agentRuns.id)) + .where(and( + eq(agentRuns.taskId, taskId), + eq(artifacts.artifactType, 'adr_text'), + )) + .orderBy(desc(artifacts.createdAt)) + .limit(1) + const metadata = artifact && isRecord(artifact.metadata) ? artifact.metadata : null + const planVersion = metadata?.planVersion + const envelope = artifact && typeof planVersion === 'string' + ? readLegacyClarification(metadata, { + taskId, + agentRunId: artifact.agentRunId, + planVersion, + }) + : null + const summaryById = new Map(summaries.map((summary) => [summary.id, summary])) + if (!envelope + || envelope.questions.length !== summaries.length + || envelope.questions.some((question) => !summaryById.has(question.id))) { + return NextResponse.json({ error: 'Legacy clarification history is unavailable.' }, { status: 409 }) + } + return NextResponse.json({ + questions: envelope.questions.map((question) => { + const summary = summaryById.get(question.id)! + return { + ...summary, + question: question.question, + suggestions: question.suggestions, + answer: summary.status === 'answered' ? question.answer : null, + } + }), + }) } catch { console.error('[GET /api/tasks/:id/questions] Unexpected error') return NextResponse.json({ error: 'Internal server error' }, { status: 500 }) @@ -118,42 +189,194 @@ export async function POST( const { answers } = parsed.data const questionIds = answers.map((a) => a.id) + const storage = architectPlanStorageConfiguration(process.env, await readS4RuntimeModeV1()) + let result: + | { + ok: true + updatedQuestions: Array<{ + id: string + status: string + createdAt: Date + answeredAt: Date | null + }> + allAnswered: boolean + } + | { ok: false; error: string; status: 400 | 409 } - const existingQuestions = await db - .select({ id: taskQuestions.id, sourcePlanArtifactId: taskQuestions.sourcePlanArtifactId, sourcePlanVersion: taskQuestions.sourcePlanVersion }) - .from(taskQuestions) - .where(and(eq(taskQuestions.taskId, taskId), inArray(taskQuestions.id, questionIds))) + if (storage.mode === 'legacy') { + result = await db.transaction(async (tx) => { + const existingQuestions = await tx + .select({ + id: taskQuestions.id, + status: taskQuestions.status, + createdAt: taskQuestions.createdAt, + answeredAt: taskQuestions.answeredAt, + answerReferenceId: taskQuestions.answerReferenceId, + questionEntryId: taskQuestions.questionEntryId, + sourcePlanArtifactId: taskQuestions.sourcePlanArtifactId, + sourcePlanVersion: taskQuestions.sourcePlanVersion, + }) + .from(taskQuestions) + .where(eq(taskQuestions.taskId, taskId)) + .for('update') + const existingIds = new Set(existingQuestions.map((question) => question.id)) + const unknownIds = questionIds.filter((id) => !existingIds.has(id)) + if (unknownIds.length > 0) { + return { + ok: false as const, + error: `Unknown question id(s) for this task: ${unknownIds.join(', ')}`, + status: 400 as const, + } + } + if (existingQuestions.some((question) => + question.questionEntryId !== null + || question.sourcePlanArtifactId !== null + || question.sourcePlanVersion !== null + || question.answerReferenceId !== null)) { + return { + ok: false as const, + error: 'Legacy clarification state is not answerable.', + status: 409 as const, + } + } - const existingIds = new Set(existingQuestions.map((q) => q.id)) - const unknownIds = questionIds.filter((id) => !existingIds.has(id)) - if (unknownIds.length > 0) { - return NextResponse.json( - { error: `Unknown question id(s) for this task: ${unknownIds.join(', ')}` }, - { status: 400 }, - ) - } + const [artifact] = await tx + .select(legacyArtifactSelection) + .from(artifacts) + .innerJoin(agentRuns, eq(artifacts.agentRunId, agentRuns.id)) + .where(and( + eq(agentRuns.taskId, taskId), + eq(artifacts.artifactType, 'adr_text'), + )) + .orderBy(desc(artifacts.createdAt)) + .limit(1) + .for('update') + const metadata = artifact && isRecord(artifact.metadata) ? artifact.metadata : null + const planVersion = metadata?.planVersion + const envelope = artifact && typeof planVersion === 'string' + ? readLegacyClarification(metadata, { + taskId, + agentRunId: artifact.agentRunId, + planVersion, + }) + : null + const answeredEnvelope = envelope + ? answerLegacyClarification(envelope, answers) + : null + if (!artifact || !metadata || !answeredEnvelope) { + return { + ok: false as const, + error: 'Legacy clarification history is unavailable.', + status: 409 as const, + } + } + const envelopeQuestionById = new Map(envelope!.questions.map((question) => [question.id, question])) + if (envelopeQuestionById.size !== existingQuestions.length + || existingQuestions.some((question) => { + const durableQuestion = envelopeQuestionById.get(question.id) + if (!durableQuestion) return true + if (durableQuestion.answer === null) { + return question.status !== 'open' || question.answeredAt !== null + } + return question.status !== 'answered' || question.answeredAt === null + }) + || questionIds.some((id) => envelopeQuestionById.get(id)?.answer !== null)) { + return { + ok: false as const, + error: 'Legacy clarification projection does not match its durable artifact.', + status: 409 as const, + } + } - const credential = readSessionCredential(request) - const storage = architectPlanStorageConfiguration(process.env, await readS4RuntimeModeV1()) - if (!credential || storage.mode !== 'protected') { - return NextResponse.json({ error: 'Protected clarification history is unavailable.' }, { status: 409 }) - } - const sourceById = new Map(existingQuestions.map((question) => [question.id, question])) - if ([...sourceById.values()].some((question) => !question.sourcePlanArtifactId || !question.sourcePlanVersion)) { - return NextResponse.json({ error: 'Clarification source is unavailable.' }, { status: 409 }) + const now = new Date() + const [updatedArtifact] = await tx + .update(artifacts) + .set({ + metadata: { + ...metadata, + [LEGACY_CLARIFICATION_METADATA_KEY]: sealLegacyClarification(answeredEnvelope), + }, + }) + .where(eq(artifacts.id, artifact.id)) + .returning({ id: artifacts.id }) + if (!updatedArtifact) throw new Error('Legacy clarification artifact update failed.') + const updatedQuestions = await tx + .update(taskQuestions) + .set({ + status: 'answered', + answeredAt: now, + answeredBy: session.userId, + }) + .where(and(eq(taskQuestions.taskId, taskId), inArray(taskQuestions.id, questionIds))) + .returning({ + id: taskQuestions.id, + status: taskQuestions.status, + createdAt: taskQuestions.createdAt, + answeredAt: taskQuestions.answeredAt, + }) + if (updatedQuestions.length !== questionIds.length) { + throw new Error('Legacy clarification projection update failed.') + } + return { + ok: true as const, + updatedQuestions, + allAnswered: legacyClarificationAllAnswered(answeredEnvelope), + } + }) + } else { + const existingQuestions = await db + .select({ + id: taskQuestions.id, + sourcePlanArtifactId: taskQuestions.sourcePlanArtifactId, + sourcePlanVersion: taskQuestions.sourcePlanVersion, + }) + .from(taskQuestions) + .where(and(eq(taskQuestions.taskId, taskId), inArray(taskQuestions.id, questionIds))) + const existingIds = new Set(existingQuestions.map((question) => question.id)) + const unknownIds = questionIds.filter((id) => !existingIds.has(id)) + if (unknownIds.length > 0) { + return NextResponse.json( + { error: `Unknown question id(s) for this task: ${unknownIds.join(', ')}` }, + { status: 400 }, + ) + } + const credential = readSessionCredential(request) + if (!credential) { + return NextResponse.json({ error: 'Protected clarification history is unavailable.' }, { status: 409 }) + } + const sourceById = new Map(existingQuestions.map((question) => [question.id, question])) + if ([...sourceById.values()].some((question) => !question.sourcePlanArtifactId || !question.sourcePlanVersion)) { + return NextResponse.json({ error: 'Clarification source is unavailable.' }, { status: 409 }) + } + const appended = [] + for (const answer of answers) { + const source = sourceById.get(answer.id)! + appended.push(await appendArchitectClarificationAnswer({ + answer: answer.answer, digestKey: storage.digestKey, digestKeyId: storage.digestKeyId, + questionId: answer.id, sessionCredential: credential, + sourcePlanArtifactId: source.sourcePlanArtifactId!, sourcePlanVersion: String(source.sourcePlanVersion), taskId, + })) + } + const updatedQuestions = await db + .select({ + id: taskQuestions.id, + status: taskQuestions.status, + createdAt: taskQuestions.createdAt, + answeredAt: taskQuestions.answeredAt, + }) + .from(taskQuestions) + .where(and(eq(taskQuestions.taskId, taskId), inArray(taskQuestions.id, questionIds))) + result = { + ok: true, + updatedQuestions, + allAnswered: appended.at(-1)?.allAnswered === true, + } } - const appended = [] - for (const answer of answers) { - const source = sourceById.get(answer.id)! - appended.push(await appendArchitectClarificationAnswer({ - answer: answer.answer, digestKey: storage.digestKey, digestKeyId: storage.digestKeyId, - questionId: answer.id, sessionCredential: credential, - sourcePlanArtifactId: source.sourcePlanArtifactId!, sourcePlanVersion: String(source.sourcePlanVersion), taskId, - })) + + if (!result.ok) { + return NextResponse.json({ error: result.error }, { status: result.status }) } - const updatedQuestions = await db.select({ id: taskQuestions.id, status: taskQuestions.status, createdAt: taskQuestions.createdAt, answeredAt: taskQuestions.answeredAt }) - .from(taskQuestions).where(and(eq(taskQuestions.taskId, taskId), inArray(taskQuestions.id, questionIds))) - const allAnswered = appended.at(-1)?.allAnswered === true + const { updatedQuestions, allAnswered } = result await publishTaskEvent(taskId, 'questions:answered', { answeredCount: updatedQuestions.length, diff --git a/web/app/dashboard/tasks/[id]/page.tsx b/web/app/dashboard/tasks/[id]/page.tsx index d96deaef..5711ffe6 100644 --- a/web/app/dashboard/tasks/[id]/page.tsx +++ b/web/app/dashboard/tasks/[id]/page.tsx @@ -4256,7 +4256,14 @@ export default function TaskDetailPage() { summaries: TaskQuestionSummary[], ) => { if (!planVersion) { - setClarificationQuestions([]) + try { + const response = await fetch(`/api/tasks/${taskId}/questions`) + if (!response.ok) throw new Error('Legacy clarification history is unavailable') + const body = await response.json() as { questions?: TaskQuestion[] } + setClarificationQuestions(body.questions ?? []) + } catch { + setClarificationQuestions([]) + } return } try { diff --git a/web/db/migrations/0027_epic_172_s4_packet_context.sql b/web/db/migrations/0027_epic_172_s4_packet_context.sql index 546cce36..2daf6655 100644 --- a/web/db/migrations/0027_epic_172_s4_packet_context.sql +++ b/web/db/migrations/0027_epic_172_s4_packet_context.sql @@ -7304,7 +7304,10 @@ ALTER TABLE public.task_questions (question_entry_id IS NOT NULL AND source_plan_artifact_id IS NOT NULL AND source_plan_version IS NOT NULL AND ((status = 'open' AND answer_reference_id IS NULL) OR (status = 'answered' AND answer_reference_id IS NOT NULL))) OR (question_entry_id IS NULL AND source_plan_artifact_id IS NULL AND source_plan_version IS NULL - AND answer_reference_id IS NULL AND status = 'legacy_unavailable') + AND answer_reference_id IS NULL AND ( + (status IN ('legacy_unavailable', 'open') AND answered_at IS NULL) + OR (status = 'answered' AND answered_at IS NOT NULL) + )) ); CREATE TABLE public.architect_clarification_answer_writes ( id uuid PRIMARY KEY DEFAULT pg_catalog.gen_random_uuid(), diff --git a/web/db/schema.ts b/web/db/schema.ts index 396f5e72..211a71d5 100644 --- a/web/db/schema.ts +++ b/web/db/schema.ts @@ -2323,7 +2323,7 @@ export const taskQuestions = pgTable( sourcePlanArtifactId: uuid('source_plan_artifact_id'), sourcePlanVersion: bigint('source_plan_version', { mode: 'number' }), answerReferenceId: uuid('answer_reference_id'), - // 'open'|'answered' + // 'open'|'answered'|'legacy_unavailable' status: text('status').notNull().default('open'), createdAt: timestamp('created_at', tsOpts).defaultNow().notNull(), answeredAt: timestamp('answered_at', tsOpts), diff --git a/web/lib/mcps/legacy-clarification.ts b/web/lib/mcps/legacy-clarification.ts new file mode 100644 index 00000000..90d151cd --- /dev/null +++ b/web/lib/mcps/legacy-clarification.ts @@ -0,0 +1,152 @@ +import { decryptSecret, encryptSecret } from '@/lib/crypto' + +export const LEGACY_CLARIFICATION_METADATA_KEY = 'legacyClarificationV1' +export const LEGACY_CLARIFICATION_MAX_TEXT_BYTES = 65_536 +const LEGACY_CLARIFICATION_MAX_QUESTIONS = 128 +const UUID_V4 = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/u +const PLAN_VERSION = /^[1-9][0-9]{0,18}$/u + +export type LegacyClarificationQuestion = Readonly<{ + id: string + question: string + suggestions: string[] + answer: string | null +}> + +export type LegacyClarificationEnvelope = Readonly<{ + schemaVersion: 1 + taskId: string + agentRunId: string + planVersion: string + questions: LegacyClarificationQuestion[] +}> + +type LegacyClarificationBinding = Readonly<{ + taskId: string + agentRunId: string + planVersion: string +}> + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function hasExactKeys(value: Record, keys: readonly string[]): boolean { + const actual = Object.keys(value).sort() + const expected = [...keys].sort() + return actual.length === expected.length && actual.every((key, index) => key === expected[index]) +} + +function boundedText(value: unknown): value is string { + return typeof value === 'string' + && value.trim() !== '' + && Buffer.byteLength(value, 'utf8') <= LEGACY_CLARIFICATION_MAX_TEXT_BYTES +} + +function canonicalQuestion(value: unknown): LegacyClarificationQuestion | null { + if (!isRecord(value) || !hasExactKeys(value, ['answer', 'id', 'question', 'suggestions'])) return null + if (typeof value.id !== 'string' || !UUID_V4.test(value.id) || !boundedText(value.question)) return null + if (!Array.isArray(value.suggestions) || value.suggestions.length > 4) return null + if (!value.suggestions.every(boundedText)) return null + if (value.answer !== null && !boundedText(value.answer)) return null + return { + id: value.id, + question: value.question, + suggestions: [...value.suggestions], + answer: value.answer, + } +} + +function canonicalEnvelope(value: unknown): LegacyClarificationEnvelope | null { + if (!isRecord(value) + || !hasExactKeys(value, ['agentRunId', 'planVersion', 'questions', 'schemaVersion', 'taskId']) + || value.schemaVersion !== 1 + || typeof value.taskId !== 'string' + || typeof value.agentRunId !== 'string' + || !UUID_V4.test(value.taskId) + || !UUID_V4.test(value.agentRunId) + || typeof value.planVersion !== 'string' + || !PLAN_VERSION.test(value.planVersion) + || !Array.isArray(value.questions) + || value.questions.length === 0 + || value.questions.length > LEGACY_CLARIFICATION_MAX_QUESTIONS) { + return null + } + const questions = value.questions.map(canonicalQuestion) + if (questions.some((question) => question === null)) return null + const ids = new Set(questions.map((question) => question!.id)) + if (ids.size !== questions.length) return null + return { + schemaVersion: 1, + taskId: value.taskId, + agentRunId: value.agentRunId, + planVersion: value.planVersion, + questions: questions as LegacyClarificationQuestion[], + } +} + +function matchesBinding( + envelope: LegacyClarificationEnvelope, + binding: LegacyClarificationBinding, +): boolean { + return envelope.taskId === binding.taskId + && envelope.agentRunId === binding.agentRunId + && envelope.planVersion === binding.planVersion +} + +export function sealLegacyClarification( + input: LegacyClarificationEnvelope, +): Record { + const envelope = canonicalEnvelope(input) + if (!envelope) throw new Error('Legacy clarification data is invalid.') + return { + schemaVersion: 1, + ciphertext: encryptSecret(JSON.stringify(envelope)), + } +} + +export function readLegacyClarification( + metadata: unknown, + binding: LegacyClarificationBinding, +): LegacyClarificationEnvelope | null { + if (!isRecord(metadata)) return null + const sealed = metadata[LEGACY_CLARIFICATION_METADATA_KEY] + if (!isRecord(sealed) + || !hasExactKeys(sealed, ['ciphertext', 'schemaVersion']) + || sealed.schemaVersion !== 1 + || typeof sealed.ciphertext !== 'string') { + return null + } + try { + const envelope = canonicalEnvelope(JSON.parse(decryptSecret(sealed.ciphertext))) + return envelope && matchesBinding(envelope, binding) ? envelope : null + } catch { + return null + } +} + +export function answerLegacyClarification( + envelope: LegacyClarificationEnvelope, + answers: readonly Readonly<{ id: string; answer: string }>[], +): LegacyClarificationEnvelope | null { + if (answers.length === 0) return null + const answerById = new Map() + for (const item of answers) { + if (!UUID_V4.test(item.id) || !boundedText(item.answer) || answerById.has(item.id)) return null + answerById.set(item.id, item.answer) + } + const known = new Set(envelope.questions.map((question) => question.id)) + if ([...answerById.keys()].some((id) => !known.has(id))) return null + return { + ...envelope, + questions: envelope.questions.map((question) => { + const answer = answerById.get(question.id) + return answer === undefined ? question : { ...question, answer } + }), + } +} + +export function legacyClarificationAllAnswered(envelope: LegacyClarificationEnvelope): boolean { + return envelope.questions.length > 0 + && envelope.questions.every((question) => question.answer !== null) +} diff --git a/web/worker/orchestrator.ts b/web/worker/orchestrator.ts index 6585f399..8d521e57 100644 --- a/web/worker/orchestrator.ts +++ b/web/worker/orchestrator.ts @@ -52,6 +52,12 @@ import { import { ARCHITECT_PLAN_HEADER, } from '../lib/mcps/architect-plan-entries' +import { + LEGACY_CLARIFICATION_METADATA_KEY, + legacyClarificationAllAnswered, + readLegacyClarification, + sealLegacyClarification, +} from '../lib/mcps/legacy-clarification' import { readS4RuntimeModeV1 } from '../lib/mcps/s4-lease' import { appendProtectedArchitectClarifications, @@ -460,6 +466,7 @@ function mockArchitectPlan(task: TaskRow, project: ProjectRow): string { export type LatestPlanArtifact = { id?: string + agentRunId?: string content: string metadata: Record } @@ -528,7 +535,12 @@ function regeneratedPlanText(planText: string): string { async function loadLatestPlanArtifact(taskId: string): Promise { const [artifact] = await db - .select({ id: artifacts.id, content: artifacts.content, metadata: artifacts.metadata }) + .select({ + id: artifacts.id, + agentRunId: artifacts.agentRunId, + content: artifacts.content, + metadata: artifacts.metadata, + }) .from(artifacts) .innerJoin(agentRuns, eq(artifacts.agentRunId, agentRuns.id)) .where(and(eq(agentRuns.taskId, taskId), eq(artifacts.artifactType, 'adr_text'))) @@ -538,6 +550,7 @@ async function loadLatestPlanArtifact(taskId: string): Promise, + claimLeaseFence: ClaimLeaseFence, +): Promise { + claimLeaseFence.assertOwned() + const artifact = await loadLatestPlanArtifact(taskId) + claimLeaseFence.assertOwned() + const planVersion = artifact?.metadata.planVersion + const envelope = artifact + && artifact.agentRunId + && typeof planVersion === 'string' + ? readLegacyClarification(artifact.metadata, { + taskId, + agentRunId: artifact.agentRunId, + planVersion, + }) + : null + if (!envelope || !legacyClarificationAllAnswered(envelope)) { + throw new Error('Legacy clarification history is unavailable for re-plan.') + } + const rowById = new Map(questions.map((question) => [question.id, question])) + if (rowById.size !== envelope.questions.length + || envelope.questions.some((question) => { + const row = rowById.get(question.id) + return !row + || row.status !== 'answered' + || row.answeredAt === null + || row.questionEntryId !== null + || row.sourcePlanArtifactId !== null + || row.sourcePlanVersion !== null + || row.answerReferenceId !== null + })) { + throw new Error('Legacy clarification projection does not match its durable artifact.') + } + return envelope.questions.map((question) => ({ + questionId: question.id, + answerId: question.id, + question: question.question, + answer: question.answer!, + })) +} + async function runArchitect( task: TaskRow, project: ProjectRow, @@ -1196,6 +1252,22 @@ async function runArchitect( // Revised plans carry only their current protected open-question entries. answeredQuestions: [], }) + const legacyClarificationMetadata = s4RuntimeMode === 'legacy' && protectedOpenQuestions.length > 0 + ? { + [LEGACY_CLARIFICATION_METADATA_KEY]: sealLegacyClarification({ + schemaVersion: 1, + taskId: task.id, + agentRunId: run.id, + planVersion, + questions: protectedOpenQuestions.map((question) => ({ + id: question.questionId, + question: question.question, + suggestions: question.suggestions, + answer: null, + })), + }), + } + : {} claimLeaseFence.assertOwned() const artifact = await createArchitectPlanArtifact(task.id, run.id, artifactPlanText, planVersion, { openQuestionCount: prepared.questions.length, @@ -1211,6 +1283,7 @@ async function runArchitect( mcpExecutionDesign: previousPlan !== null && artifactComparableMetadata === previousComparableMetadata && isRecord(previousPlanArtifact?.metadata.mcpExecutionDesign) ? previousPlanArtifact.metadata.mcpExecutionDesign : prepared.mcpExecutionDesign, + ...legacyClarificationMetadata, }, protectedEntries, claimLeaseFence) claimLeaseFence.assertOwned() const openQuestionCount = await persistOpenQuestions( @@ -1298,7 +1371,10 @@ async function runArchitect( runStatus: 'completed', artifactId: artifact.id, openQuestionCount, - openQuestions: prepared.questions.map((question) => question.question), + // Question text remains only in protected history or the encrypted + // legacy artifact envelope. Checkpoints retain the count, not a derived + // plaintext copy. + openQuestions: [], revisedFromAnswers: answeredQuestions.length > 0, revisedFromPlan: previousPlan !== null, protectedHistory: isRecord(artifact.metadata) && artifact.metadata.historyAvailable === true, @@ -1537,7 +1613,12 @@ export async function processAnsweredQuestions( return 'completed' } - const answeredQuestions = answeredQuestionSnapshot(existingQuestions) + const runtimeMode = await readS4RuntimeModeV1() + claimLeaseFence.assertOwned() + const storage = architectPlanStorageConfiguration(process.env, runtimeMode) + const answeredQuestions = storage.mode === 'legacy' + ? await legacyAnsweredQuestionSnapshot(taskId, existingQuestions, claimLeaseFence) + : answeredQuestionSnapshot(existingQuestions) if (!recoveredRunningOccurrence) { claimLeaseFence.assertOwned() From aa709ce78d72aff1802ea83d6f7ca8b98eb8a6c4 Mon Sep 17 00:00:00 2001 From: Joncallim <64296013+Joncallim@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:33:09 +0800 Subject: [PATCH 3/8] fix: certify clarification S4 objects --- web/__tests__/epic-172-s4-context.test.ts | 31 ++++++++++++++++++++++ web/scripts/bootstrap-epic-172-s4-roles.ts | 26 ++++++++++++++++-- 2 files changed, 55 insertions(+), 2 deletions(-) diff --git a/web/__tests__/epic-172-s4-context.test.ts b/web/__tests__/epic-172-s4-context.test.ts index e98eca15..44eefade 100644 --- a/web/__tests__/epic-172-s4-context.test.ts +++ b/web/__tests__/epic-172-s4-context.test.ts @@ -278,6 +278,37 @@ describe('Epic 172 S4 PostgreSQL CI contract', () => { expect(s4Migration).toMatch(/RETURNS TABLE \(purpose text, source_kind text, task_id uuid/) }) + it('certifies every protected clarification table and routine in the S4 owner finalizer', () => { + const ownedTableInventory = s4RoleBootstrap.match( + /const OWNED_TABLES = \[([\s\S]*?)\] as const/, + )?.[1] ?? '' + const routineInventory = s4RoleBootstrap.match( + /routine\.proname = any\(array\[([\s\S]*?)\]\)\s+and routine\.proowner/, + )?.[1] ?? '' + + for (const table of [ + 'architect_clarification_answers', + 'architect_clarification_answer_writes', + ]) { + expect(ownedTableInventory).toContain(`'${table}'`) + } + for (const routine of [ + 'bind_architect_replan_context_v3', + 'resolve_architect_plan_entry_v2', + 'append_architect_clarification_answer_v1', + ]) { + expect(routineInventory).toContain(`'${routine}'`) + } + + expect(s4RoleBootstrap).toContain('acl.grantee <> table_row.relowner') + expect(s4RoleBootstrap).toContain("acl.grantee = 0 and acl.privilege_type = 'EXECUTE'") + expect(s4RoleBootstrap).toContain("when 'bind_architect_replan_context_v3'\n then 'forge_architect_plan_writer'::regrole") + expect(s4RoleBootstrap).toContain("when 'resolve_architect_plan_entry_v2'\n then 'forge_architect_plan_resolver'::regrole") + expect(s4RoleBootstrap).toContain("when 'append_architect_clarification_answer_v1'\n then 'forge_architect_plan_history_reader'::regrole") + expect(s4RoleBootstrap).toContain('acl.grantee <> case routine.proname') + expect(s4RoleBootstrap).toContain(') <> 73 then') + }) + it('audits the complete protected clarification history set without truncation', () => { const historyReader = s4Migration.match( /CREATE OR REPLACE FUNCTION forge\.read_architect_plan_history_v1\([\s\S]*?\n\$\$;/, diff --git a/web/scripts/bootstrap-epic-172-s4-roles.ts b/web/scripts/bootstrap-epic-172-s4-roles.ts index 5f2f32a6..8504c8f8 100644 --- a/web/scripts/bootstrap-epic-172-s4-roles.ts +++ b/web/scripts/bootstrap-epic-172-s4-roles.ts @@ -19,6 +19,8 @@ const OWNED_TABLES = [ 'architect_plan_entries', 'architect_plan_execution_references', 'architect_plan_history_reads', + 'architect_clarification_answers', + 'architect_clarification_answer_writes', 'protected_package_entry_registrations', 'protected_entry_capability_bindings', 'mcp_operator_review_versions', @@ -436,6 +438,9 @@ async function main(): Promise { ,'apply_local_effect_recovery_action_v2' ,'apply_packet_issuance_recovery_action_v2' ,'bind_architect_replan_context_v2' + ,'bind_architect_replan_context_v3' + ,'resolve_architect_plan_entry_v2' + ,'append_architect_clarification_answer_v1' ,'local_projection_archive_operation_fingerprint_v2' ,'inspect_local_projection_overlimit_v2' ,'apply_local_projection_overlimit_archive_v2' @@ -452,9 +457,26 @@ async function main(): Promise { pg_catalog.acldefault('f', routine.proowner) ) ) acl - where acl.grantee = 0 and acl.privilege_type = 'EXECUTE' + where (acl.grantee = 0 and acl.privilege_type = 'EXECUTE') + or ( + routine.proname = any(array[ + 'bind_architect_replan_context_v3', + 'resolve_architect_plan_entry_v2', + 'append_architect_clarification_answer_v1' + ]) + and acl.privilege_type = 'EXECUTE' + and acl.grantee <> routine.proowner + and acl.grantee <> case routine.proname + when 'bind_architect_replan_context_v3' + then 'forge_architect_plan_writer'::regrole + when 'resolve_architect_plan_entry_v2' + then 'forge_architect_plan_resolver'::regrole + when 'append_architect_clarification_answer_v1' + then 'forge_architect_plan_history_reader'::regrole + end + ) ) - ) <> 70 then + ) <> 73 then raise exception 'The S4 routine owner or PUBLIC boundary is incomplete' using errcode = '42501'; end if; From 10c2154b33f5d4fe55597be54a47eeefa3c609d4 Mon Sep 17 00:00:00 2001 From: Joncallim <64296013+Joncallim@users.noreply.github.com> Date: Thu, 30 Jul 2026 23:38:39 +0800 Subject: [PATCH 4/8] fix: certify exact clarification routine authority --- web/__tests__/epic-172-s4-context.test.ts | 48 ++++++++-- web/__tests__/epic-172-s4-postgres.test.ts | 86 +++++++++++++++++ web/scripts/bootstrap-epic-172-s4-roles.ts | 103 +++++++++++++++++---- 3 files changed, 209 insertions(+), 28 deletions(-) diff --git a/web/__tests__/epic-172-s4-context.test.ts b/web/__tests__/epic-172-s4-context.test.ts index 44eefade..8dd242b6 100644 --- a/web/__tests__/epic-172-s4-context.test.ts +++ b/web/__tests__/epic-172-s4-context.test.ts @@ -282,8 +282,8 @@ describe('Epic 172 S4 PostgreSQL CI contract', () => { const ownedTableInventory = s4RoleBootstrap.match( /const OWNED_TABLES = \[([\s\S]*?)\] as const/, )?.[1] ?? '' - const routineInventory = s4RoleBootstrap.match( - /routine\.proname = any\(array\[([\s\S]*?)\]\)\s+and routine\.proowner/, + const exactRoutineInventory = s4RoleBootstrap.match( + /const EXACT_CLARIFICATION_ROUTINES = \[([\s\S]*?)\] as const/, )?.[1] ?? '' for (const table of [ @@ -293,19 +293,47 @@ describe('Epic 172 S4 PostgreSQL CI contract', () => { expect(ownedTableInventory).toContain(`'${table}'`) } for (const routine of [ - 'bind_architect_replan_context_v3', - 'resolve_architect_plan_entry_v2', - 'append_architect_clarification_answer_v1', + { + identity: 'forge.bind_architect_replan_context_v3(uuid,uuid)', + name: 'bind_architect_replan_context_v3', + grantee: 'forge_architect_plan_writer', + }, + { + identity: 'forge.resolve_architect_plan_entry_v2(uuid)', + name: 'resolve_architect_plan_entry_v2', + grantee: 'forge_architect_plan_resolver', + }, + { + identity: 'forge.append_architect_clarification_answer_v1(bytea,uuid,uuid,uuid,bigint,uuid,text,text,text)', + name: 'append_architect_clarification_answer_v1', + grantee: 'forge_architect_plan_history_reader', + }, ]) { - expect(routineInventory).toContain(`'${routine}'`) + expect(exactRoutineInventory).toContain(`identity: '${routine.identity}'`) + expect(exactRoutineInventory).toContain(`name: '${routine.name}'`) + expect(exactRoutineInventory).toContain(`grantee: '${routine.grantee}'`) } expect(s4RoleBootstrap).toContain('acl.grantee <> table_row.relowner') expect(s4RoleBootstrap).toContain("acl.grantee = 0 and acl.privilege_type = 'EXECUTE'") - expect(s4RoleBootstrap).toContain("when 'bind_architect_replan_context_v3'\n then 'forge_architect_plan_writer'::regrole") - expect(s4RoleBootstrap).toContain("when 'resolve_architect_plan_entry_v2'\n then 'forge_architect_plan_resolver'::regrole") - expect(s4RoleBootstrap).toContain("when 'append_architect_clarification_answer_v1'\n then 'forge_architect_plan_history_reader'::regrole") - expect(s4RoleBootstrap).toContain('acl.grantee <> case routine.proname') + expect(s4RoleBootstrap).toContain( + 'routine.oid = pg_catalog.to_regprocedure(expected.routine_identity)', + ) + expect(s4RoleBootstrap).toMatch( + /if exists \(\s+with expected\(routine_identity, routine_name, expected_grantee\)/, + ) + expect(s4RoleBootstrap).toContain('observed.proowner <>') + expect(s4RoleBootstrap).toContain('observed.acl_count <> 2') + expect(s4RoleBootstrap).toContain('observed.owner_execute_count <> 1') + expect(s4RoleBootstrap).toContain('observed.expected_execute_count <> 1') + expect(s4RoleBootstrap).toContain('and not acl.is_grantable') + expect(s4RoleBootstrap).toContain( + 'pg_catalog.to_regprocedure(expected.routine_identity) = routine.oid', + ) + expect(s4RoleBootstrap).toContain( + "raise exception 'The exact S4 clarification routine authority is incomplete'", + ) + expect(s4RoleBootstrap).not.toContain('acl.grantee <> case routine.proname') expect(s4RoleBootstrap).toContain(') <> 73 then') }) diff --git a/web/__tests__/epic-172-s4-postgres.test.ts b/web/__tests__/epic-172-s4-postgres.test.ts index dd3a0383..da614acd 100644 --- a/web/__tests__/epic-172-s4-postgres.test.ts +++ b/web/__tests__/epic-172-s4-postgres.test.ts @@ -1484,6 +1484,92 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { expect(row).toEqual({ agentRunId: runId, state: 'claimed' }) }) + it('rejects hostile clarification routine identities and ACL tuples without retaining mutations', async () => { + const rollbackMarker = 'S4 clarification routine authority probe rollback' + const authorityError = 'The exact S4 clarification routine authority is incomplete' + + async function runAuthorityProbe(mutation: string): Promise<'accepted' | 'rejected'> { + try { + await admin.begin(async (tx) => { + const [{ migrationRole }] = await tx<{ migrationRole: string }[]>` + select database_row.datdba::pg_catalog.regrole::text as "migrationRole" + from pg_catalog.pg_database database_row + where database_row.datname = pg_catalog.current_database() + ` + await tx.unsafe(` + alter role forge_s4_routines_owner password null; + alter role forge_architect_plan_writer password null; + alter role forge_architect_plan_resolver password null; + alter role forge_architect_plan_history_reader password null; + alter role forge_packet_issuer password null; + alter role forge_review_source_resolver password null; + alter role forge_s4_recovery_operator password null; + alter role forge_local_projection_archiver password null; + alter role forge_project_root_reconciler password null; + `) + await tx`grant forge_s4_routines_owner to ${tx(migrationRole)} + with admin false, inherit false, set true` + await tx`grant execute on function + public.forge_finalize_epic_172_s4_owner_bootstrap_v1() + to ${tx(migrationRole)}` + await tx.unsafe(mutation) + await tx`set local session authorization ${tx(migrationRole)}` + await tx`select public.forge_finalize_epic_172_s4_owner_bootstrap_v1()` + throw new Error(rollbackMarker) + }) + } catch (error) { + if (error instanceof Error && error.message === rollbackMarker) return 'accepted' + if ( + typeof error === 'object' + && error !== null + && 'code' in error + && error.code === '42501' + && 'message' in error + && error.message === authorityError + ) { + return 'rejected' + } + throw new Error('The S4 clarification routine authority probe failed unexpectedly.') + } + throw new Error('The S4 clarification routine authority probe did not roll back.') + } + + const hostileMutations = [ + ` + grant execute on function forge.bind_architect_replan_context_v3(uuid,uuid) + to forge_packet_issuer; + `, + ` + grant execute on function forge.resolve_architect_plan_entry_v2(uuid) + to forge_architect_plan_resolver with grant option; + `, + ` + revoke execute on function + forge.append_architect_clarification_answer_v1( + bytea,uuid,uuid,uuid,bigint,uuid,text,text,text + ) + from forge_architect_plan_history_reader; + `, + ` + alter function forge.resolve_architect_plan_entry_v2(uuid) + rename to resolve_architect_plan_entry_v2_exact_probe; + create function forge.resolve_architect_plan_entry_v2(text) + returns void language plpgsql as 'begin return; end'; + revoke all on function forge.resolve_architect_plan_entry_v2(text) from public; + alter function forge.resolve_architect_plan_entry_v2(text) + owner to forge_s4_routines_owner; + grant execute on function forge.resolve_architect_plan_entry_v2(text) + to forge_architect_plan_resolver; + `, + ] + + expect(await runAuthorityProbe('')).toBe('accepted') + for (const mutation of hostileMutations) { + expect(await runAuthorityProbe(mutation)).toBe('rejected') + expect(await runAuthorityProbe('')).toBe('accepted') + } + }) + }) describe.skipIf(!enabled)('Epic 172 legacy leakage scrub PostgreSQL proof', () => { diff --git a/web/scripts/bootstrap-epic-172-s4-roles.ts b/web/scripts/bootstrap-epic-172-s4-roles.ts index 8504c8f8..cd01d3fc 100644 --- a/web/scripts/bootstrap-epic-172-s4-roles.ts +++ b/web/scripts/bootstrap-epic-172-s4-roles.ts @@ -43,6 +43,23 @@ const OWNED_TABLES = [ 'local_projection_archive_operations', 'local_projection_archive_operation_checkpoints', ] as const +const EXACT_CLARIFICATION_ROUTINES = [ + { + identity: 'forge.bind_architect_replan_context_v3(uuid,uuid)', + name: 'bind_architect_replan_context_v3', + grantee: 'forge_architect_plan_writer', + }, + { + identity: 'forge.resolve_architect_plan_entry_v2(uuid)', + name: 'resolve_architect_plan_entry_v2', + grantee: 'forge_architect_plan_resolver', + }, + { + identity: 'forge.append_architect_clarification_answer_v1(bytea,uuid,uuid,uuid,bigint,uuid,text,text,text)', + name: 'append_architect_clarification_answer_v1', + grantee: 'forge_architect_plan_history_reader', + }, +] as const function literal(value: string): string { return `'${value.replaceAll("'", "''")}'` @@ -205,6 +222,11 @@ async function main(): Promise { } const migrationLiteral = literal(migrationRole) const tableList = OWNED_TABLES.map(literal).join(',') + const exactClarificationRoutineValues = EXACT_CLARIFICATION_ROUTINES.map((routine) => `( + ${literal(routine.identity)}, + ${literal(routine.name)}, + ${literal(routine.grantee)}::pg_catalog.regrole + )`).join(',') await admin.unsafe(` create or replace function public.forge_begin_epic_172_s4_owner_bootstrap_v1() returns void @@ -367,6 +389,68 @@ async function main(): Promise { and table_row.relname = any(array[${tableList}]) ) using errcode = '42501'; end if; + if exists ( + with expected(routine_identity, routine_name, expected_grantee) as ( + values ${exactClarificationRoutineValues} + ), + observed as ( + select + expected.routine_identity, + expected.routine_name, + expected.expected_grantee, + routine.oid as routine_oid, + routine.proowner, + pg_catalog.count(acl.grantee) as acl_count, + pg_catalog.count(acl.grantee) filter ( + where acl.grantee = routine.proowner + and acl.privilege_type = 'EXECUTE' + and not acl.is_grantable + ) as owner_execute_count, + pg_catalog.count(acl.grantee) filter ( + where acl.grantee = expected.expected_grantee + and acl.privilege_type = 'EXECUTE' + and not acl.is_grantable + ) as expected_execute_count + from expected + left join pg_catalog.pg_proc routine + on routine.oid = pg_catalog.to_regprocedure(expected.routine_identity) + left join lateral pg_catalog.aclexplode( + coalesce( + routine.proacl, + pg_catalog.acldefault('f', routine.proowner) + ) + ) acl on true + group by expected.routine_identity, expected.routine_name, + expected.expected_grantee, routine.oid, routine.proowner + ) + select 1 + from observed + where observed.routine_oid is null + or observed.proowner <> '${OWNER}'::regrole + or observed.acl_count <> 2 + or observed.owner_execute_count <> 1 + or observed.expected_execute_count <> 1 + ) or exists ( + with expected(routine_identity, routine_name, expected_grantee) as ( + values ${exactClarificationRoutineValues} + ) + select 1 + from pg_catalog.pg_proc routine + join pg_catalog.pg_namespace namespace_row + on namespace_row.oid = routine.pronamespace + where namespace_row.nspname = 'forge' + and exists ( + select 1 from expected + where expected.routine_name = routine.proname + ) + and not exists ( + select 1 from expected + where pg_catalog.to_regprocedure(expected.routine_identity) = routine.oid + ) + ) then + raise exception 'The exact S4 clarification routine authority is incomplete' + using errcode = '42501'; + end if; if ( select pg_catalog.count(*) from pg_catalog.pg_proc routine @@ -457,24 +541,7 @@ async function main(): Promise { pg_catalog.acldefault('f', routine.proowner) ) ) acl - where (acl.grantee = 0 and acl.privilege_type = 'EXECUTE') - or ( - routine.proname = any(array[ - 'bind_architect_replan_context_v3', - 'resolve_architect_plan_entry_v2', - 'append_architect_clarification_answer_v1' - ]) - and acl.privilege_type = 'EXECUTE' - and acl.grantee <> routine.proowner - and acl.grantee <> case routine.proname - when 'bind_architect_replan_context_v3' - then 'forge_architect_plan_writer'::regrole - when 'resolve_architect_plan_entry_v2' - then 'forge_architect_plan_resolver'::regrole - when 'append_architect_clarification_answer_v1' - then 'forge_architect_plan_history_reader'::regrole - end - ) + where acl.grantee = 0 and acl.privilege_type = 'EXECUTE' ) ) <> 73 then raise exception 'The S4 routine owner or PUBLIC boundary is incomplete' From 4bdd9c7197e5d801a8395d4a541572bba1cced07 Mon Sep 17 00:00:00 2001 From: Joncallim <64296013+Joncallim@users.noreply.github.com> Date: Thu, 30 Jul 2026 23:58:01 +0800 Subject: [PATCH 5/8] fix: atomically persist protected answer batches --- web/__tests__/api.test.ts | 137 +++++++++++++++++++-- web/__tests__/epic-172-s4-postgres.test.ts | 95 +++++++++++++- web/app/api/tasks/[id]/questions/route.ts | 43 +++++-- web/lib/mcps/history-reader.ts | 85 ++++++++++--- 4 files changed, 317 insertions(+), 43 deletions(-) diff --git a/web/__tests__/api.test.ts b/web/__tests__/api.test.ts index 02f5d1d3..4c23fbb5 100644 --- a/web/__tests__/api.test.ts +++ b/web/__tests__/api.test.ts @@ -17,6 +17,7 @@ import path from 'node:path' import { getTableName } from 'drizzle-orm' import { vi, describe, it, expect, beforeEach, afterEach } from 'vitest' import { canonicalS3Marker } from '../test-support/filesystem-grant-marker-fixtures' +import { LEGACY_CLARIFICATION_MAX_TEXT_BYTES } from '@/lib/mcps/legacy-clarification' const taskEventRedisEnvironment = { publisher: process.env.FORGE_TASK_EVENT_PUBLISHER_REDIS_URL, @@ -64,12 +65,12 @@ const mockLoadProtectedApprovalReviewPreflight = vi.fn().mockResolvedValue(null) const mockReadProtectedMcpOperatorReview = vi.fn().mockResolvedValue([]) const mockListApprovedPackagePlanRegistrations = vi.fn().mockResolvedValue([]) const { - mockAppendArchitectClarificationAnswer, + mockAppendArchitectClarificationAnswers, mockReadS4RuntimeModeV1, mockArchitectPlanStorageConfiguration, mockGenerateTaskTitle, } = vi.hoisted(() => ({ - mockAppendArchitectClarificationAnswer: vi.fn(), + mockAppendArchitectClarificationAnswers: vi.fn(), mockReadS4RuntimeModeV1: vi.fn().mockResolvedValue('protected'), mockArchitectPlanStorageConfiguration: vi.fn().mockReturnValue({ mode: 'protected', digestKey: Buffer.alloc(32, 7), digestKeyId: 'test-v1', @@ -82,7 +83,7 @@ vi.mock('@/lib/mcps/protected-review-preflight', () => ({ vi.mock('@/lib/mcps/history-reader', () => ({ listApprovedPackagePlanRegistrations: mockListApprovedPackagePlanRegistrations, readProtectedMcpOperatorReview: mockReadProtectedMcpOperatorReview, - appendArchitectClarificationAnswer: mockAppendArchitectClarificationAnswer, + appendArchitectClarificationAnswers: mockAppendArchitectClarificationAnswers, })) vi.mock('@/lib/mcps/s4-lease', async (importOriginal) => ({ ...await importOriginal(), @@ -7319,6 +7320,9 @@ describe('POST /api/tasks/:id/questions', () => { .mockReturnValueOnce(chain([{ id: 'task-1', status: 'awaiting_answers' }])) .mockReturnValueOnce(chain([{ id: questionId, + status: 'open', + answerReferenceId: null, + questionEntryId: `clarification_question:${questionId}`, sourcePlanArtifactId: '88888888-8888-4888-8888-888888888888', sourcePlanVersion: 1, }])) @@ -7328,7 +7332,7 @@ describe('POST /api/tasks/:id/questions', () => { createdAt: new Date('2026-07-22T00:00:00.000Z'), answeredAt: new Date('2026-07-22T00:01:00.000Z'), }])) - mockAppendArchitectClarificationAnswer.mockResolvedValue({ answerId: 'answer-1', allAnswered: true }) + mockAppendArchitectClarificationAnswers.mockResolvedValue([{ answerId: 'answer-1', allAnswered: true }]) mockRedisLpush.mockResolvedValue(1) mockRedisEval.mockResolvedValue(1) @@ -7351,9 +7355,9 @@ describe('POST /api/tasks/:id/questions', () => { allAnswered: true, }) expect(JSON.stringify(body)).not.toContain('RAW-') - expect(mockAppendArchitectClarificationAnswer).toHaveBeenCalledWith(expect.objectContaining({ - answer, questionId, taskId: 'task-1', - })) + expect(mockAppendArchitectClarificationAnswers).toHaveBeenCalledWith([ + expect.objectContaining({ answer, questionId, taskId: 'task-1' }), + ]) expect(mockDbUpdate).not.toHaveBeenCalled() expect(mockRedisPublish).not.toHaveBeenCalled() const answeredEvent = mockRedisEval.mock.calls.find((call) => call[4] === 'questions:answered') @@ -7459,7 +7463,7 @@ describe('POST /api/tasks/:id/questions', () => { suggestions: ['main'], answer, }]) - expect(mockAppendArchitectClarificationAnswer).not.toHaveBeenCalled() + expect(mockAppendArchitectClarificationAnswers).not.toHaveBeenCalled() expect(mockRedisLpush).toHaveBeenCalledWith('forge:answers', JSON.stringify({ taskId })) expect(JSON.stringify(mockRedisLpush.mock.calls)).not.toContain(answer) expect(JSON.stringify(mockRedisEval.mock.calls)).not.toContain(answer) @@ -7477,6 +7481,9 @@ describe('POST /api/tasks/:id/questions', () => { .mockReturnValueOnce(chain([{ id: taskId, status: 'awaiting_answers' }])) .mockReturnValueOnce(chain([{ id: questionId, + status: 'open', + answerReferenceId: null, + questionEntryId: `clarification_question:${questionId}`, sourcePlanArtifactId: null, sourcePlanVersion: null, }])) @@ -7491,7 +7498,119 @@ describe('POST /api/tasks/:id/questions', () => { expect(response.status).toBe(409) expect(mockDbTransaction).not.toHaveBeenCalled() expect(mockDbUpdate).not.toHaveBeenCalled() - expect(mockAppendArchitectClarificationAnswer).not.toHaveBeenCalled() + expect(mockAppendArchitectClarificationAnswers).not.toHaveBeenCalled() + }) + + it('validates the whole protected form before calling the atomic writer', async () => { + mockGetSession.mockResolvedValue(FAKE_SESSION) + const firstQuestionId = '77777777-7777-4777-8777-777777777777' + const secondQuestionId = '99999999-9999-4999-8999-999999999999' + mockDbSelect + .mockReturnValueOnce(chain([{ id: 'task-1', status: 'awaiting_answers' }])) + .mockReturnValueOnce(chain([{ + id: firstQuestionId, + status: 'open', + answerReferenceId: null, + questionEntryId: `clarification_question:${firstQuestionId}`, + sourcePlanArtifactId: '88888888-8888-4888-8888-888888888888', + sourcePlanVersion: 1, + }, { + id: secondQuestionId, + status: 'answered', + answerReferenceId: 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa', + questionEntryId: `clarification_question:${secondQuestionId}`, + sourcePlanArtifactId: '88888888-8888-4888-8888-888888888888', + sourcePlanVersion: 1, + }])) + + const { POST } = await import('@/app/api/tasks/[id]/questions/route') + const response = await POST(authRequest('/api/tasks/task-1/questions', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ answers: [ + { id: firstQuestionId, answer: 'first' }, + { id: secondQuestionId, answer: 'second' }, + ] }), + }) as never, { params: Promise.resolve({ id: 'task-1' }) }) + + expect(response.status).toBe(409) + expect(mockAppendArchitectClarificationAnswers).not.toHaveBeenCalled() + expect(mockRedisLpush).not.toHaveBeenCalled() + expect(mockRedisEval).not.toHaveBeenCalled() + }) + + it.each([ + { + name: 'duplicate ids', + answers: [ + { id: '77777777-7777-4777-8777-777777777777', answer: 'first' }, + { id: '77777777-7777-4777-8777-777777777777', answer: 'second' }, + ], + }, + { + name: 'oversized answer', + answers: [{ + id: '77777777-7777-4777-8777-777777777777', + answer: 'x'.repeat(LEGACY_CLARIFICATION_MAX_TEXT_BYTES + 1), + }], + }, + ])('rejects $name before any protected write or continuation', async ({ answers }) => { + mockGetSession.mockResolvedValue(FAKE_SESSION) + mockDbSelect.mockReturnValueOnce(chain([{ id: 'task-1', status: 'awaiting_answers' }])) + + const { POST } = await import('@/app/api/tasks/[id]/questions/route') + const response = await POST(authRequest('/api/tasks/task-1/questions', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ answers }), + }) as never, { params: Promise.resolve({ id: 'task-1' }) }) + + expect(response.status).toBe(400) + expect(mockAppendArchitectClarificationAnswers).not.toHaveBeenCalled() + expect(mockRedisLpush).not.toHaveBeenCalled() + expect(mockRedisEval).not.toHaveBeenCalled() + }) + + it('durably queues re-plan before best-effort progress publication', async () => { + mockGetSession.mockResolvedValue(FAKE_SESSION) + const questionId = '77777777-7777-4777-8777-777777777777' + mockDbSelect + .mockReturnValueOnce(chain([{ id: 'task-1', status: 'awaiting_answers' }])) + .mockReturnValueOnce(chain([{ + id: questionId, + status: 'open', + answerReferenceId: null, + questionEntryId: `clarification_question:${questionId}`, + sourcePlanArtifactId: '88888888-8888-4888-8888-888888888888', + sourcePlanVersion: 1, + }])) + .mockReturnValueOnce(chain([{ + id: questionId, + status: 'answered', + createdAt: new Date('2026-07-30T00:00:00.000Z'), + answeredAt: new Date('2026-07-30T00:01:00.000Z'), + }])) + mockAppendArchitectClarificationAnswers.mockResolvedValue([{ + answerId: 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa', + allAnswered: true, + }]) + mockRedisLpush.mockResolvedValue(1) + mockRedisEval.mockRejectedValueOnce(new Error('RAW-EVENT-OUTAGE-SENTINEL')) + + const { POST } = await import('@/app/api/tasks/[id]/questions/route') + const response = await POST(authRequest('/api/tasks/task-1/questions', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ answers: [{ id: questionId, answer: 'RAW-ANSWER-SENTINEL' }] }), + }) as never, { params: Promise.resolve({ id: 'task-1' }) }) + + expect(response.status).toBe(200) + expect(mockRedisLpush).toHaveBeenCalledTimes(1) + expect(mockRedisLpush.mock.invocationCallOrder[0]).toBeLessThan( + mockRedisEval.mock.invocationCallOrder[0], + ) + expect(JSON.stringify(mockRedisLpush.mock.calls)).not.toContain('RAW-ANSWER-SENTINEL') + expect(JSON.stringify(mockRedisEval.mock.calls)).not.toContain('RAW-ANSWER-SENTINEL') }) }) diff --git a/web/__tests__/epic-172-s4-postgres.test.ts b/web/__tests__/epic-172-s4-postgres.test.ts index da614acd..658dfb44 100644 --- a/web/__tests__/epic-172-s4-postgres.test.ts +++ b/web/__tests__/epic-172-s4-postgres.test.ts @@ -10,7 +10,11 @@ import { } from '@/lib/mcps/s4-protocol-store' import { ARCHITECT_PLAN_HEADER, architectReplanReferenceForEntry } from '@/lib/mcps/architect-plan-entries' import { computeCredentialDigest } from '@/lib/session-credential-digest' -import { appendArchitectClarificationAnswer, readArchitectPlanHistory } from '@/lib/mcps/history-reader' +import { + appendArchitectClarificationAnswer, + appendArchitectClarificationAnswers, + readArchitectPlanHistory, +} from '@/lib/mcps/history-reader' import { hashPassword } from '@/lib/password' import { closeDb } from '@/db' import { @@ -869,6 +873,95 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { await runStatefulHistoryProof() }) + it('rolls back the whole protected clarification form when a later append conflicts', async () => { + const taskId = randomUUID() + const runId = randomUUID() + const firstQuestionId = randomUUID() + const secondQuestionId = randomUUID() + const firstAnswerId = randomUUID() + const secondAnswerId = randomUUID() + await admin`insert into tasks (id, project_id, submitted_by, title, prompt, status) + values (${taskId}::uuid, ${ids.project}::uuid, ${ids.user}::uuid, + 'Atomic clarification batch', 'protected', 'awaiting_answers')` + await admin`insert into agent_runs (id, task_id, agent_type, model_id_used, status) + values (${runId}::uuid, ${taskId}::uuid, 'architect', 'test', 'completed')` + const source = await recordArchitectPlanVersion({ + agentRunId: runId, + digestKey: key, + digestKeyId: 's4-test-key', + planVersion: '1', + taskId, + entries: [ + { agent: null, bindingFingerprint: null, content: 'body', entryId: 'plan_body:000000', entryKind: 'plan_body', projectionEligible: false, requirementKey: null }, + { agent: null, bindingFingerprint: null, content: JSON.stringify({ requirementKey: 'plan-policy', schemaVersion: 1 }), entryId: 'requirement:plan-policy', entryKind: 'requirement', projectionEligible: false, requirementKey: 'plan-policy' }, + ...[firstQuestionId, secondQuestionId].map((questionId) => ({ + agent: null, + bindingFingerprint: null, + content: JSON.stringify({ + schemaVersion: 1, + questionId, + question: 'Which branch?', + suggestions: ['main'], + }), + entryId: `clarification_question:${questionId}`, + entryKind: 'clarification_question' as const, + projectionEligible: false, + requirementKey: null, + })), + ], + }) + await admin`insert into task_questions ( + id, task_id, question_entry_id, source_plan_artifact_id, source_plan_version, status + ) values + (${firstQuestionId}::uuid, ${taskId}::uuid, + ${`clarification_question:${firstQuestionId}`}, ${source.artifactId}::uuid, 1, 'open'), + (${secondQuestionId}::uuid, ${taskId}::uuid, + ${`clarification_question:${secondQuestionId}`}, ${source.artifactId}::uuid, 1, 'open')` + + const batch = [{ + answer: 'main', + answerId: firstAnswerId, + digestKey: key, + digestKeyId: 's4-test-key', + questionId: firstQuestionId, + sessionCredential, + sourcePlanArtifactId: source.artifactId, + sourcePlanVersion: '1', + taskId, + }, { + answer: 'release', + answerId: secondAnswerId, + digestKey: key, + digestKeyId: 's4-test-key', + questionId: secondQuestionId, + sessionCredential, + sourcePlanArtifactId: source.artifactId, + sourcePlanVersion: '1', + taskId, + }] + await admin`update task_questions set status = 'legacy_unavailable' + where task_id = ${taskId}::uuid and id = ${secondQuestionId}::uuid` + await expect(appendArchitectClarificationAnswers(batch)).rejects.toMatchObject({ + code: 'invalid_evidence', + }) + const [afterConflict] = await admin<{ + answerCount: number + answeredCount: number + }[]>`select + (select count(*)::integer from architect_clarification_answers + where task_id = ${taskId}::uuid) as "answerCount", + (select count(*)::integer from task_questions + where task_id = ${taskId}::uuid and status = 'answered') as "answeredCount"` + expect(afterConflict).toEqual({ answerCount: 0, answeredCount: 0 }) + + await admin`update task_questions set status = 'open' + where task_id = ${taskId}::uuid and id = ${secondQuestionId}::uuid` + await expect(appendArchitectClarificationAnswers(batch)).resolves.toEqual([ + { answerId: firstAnswerId, allAnswered: false }, + { answerId: secondAnswerId, allAnswered: true }, + ]) + }) + it('serves protected Architect history through the real password session route with PostgreSQL as authority', async () => { const ownerPassword = 'route-history-password' const routeProject = randomUUID() diff --git a/web/app/api/tasks/[id]/questions/route.ts b/web/app/api/tasks/[id]/questions/route.ts index 56a0c8ef..e6538adf 100644 --- a/web/app/api/tasks/[id]/questions/route.ts +++ b/web/app/api/tasks/[id]/questions/route.ts @@ -10,7 +10,7 @@ import { getAccessibleTask } from '@/lib/task-access' import { guardEpic172ProjectManagementIngress } from '@/lib/projects/epic-172-project-ingress' import { publishTaskEvent } from '@/worker/events' import { taskQuestionSummary } from '@/lib/mcps/clarification-projection' -import { appendArchitectClarificationAnswer } from '@/lib/mcps/history-reader' +import { appendArchitectClarificationAnswers } from '@/lib/mcps/history-reader' import { architectPlanStorageConfiguration } from '@/lib/mcps/s4-protocol-store' import { readS4RuntimeModeV1 } from '@/lib/mcps/s4-lease' import { @@ -327,11 +327,14 @@ export async function POST( const existingQuestions = await db .select({ id: taskQuestions.id, + status: taskQuestions.status, + answerReferenceId: taskQuestions.answerReferenceId, + questionEntryId: taskQuestions.questionEntryId, sourcePlanArtifactId: taskQuestions.sourcePlanArtifactId, sourcePlanVersion: taskQuestions.sourcePlanVersion, }) .from(taskQuestions) - .where(and(eq(taskQuestions.taskId, taskId), inArray(taskQuestions.id, questionIds))) + .where(eq(taskQuestions.taskId, taskId)) const existingIds = new Set(existingQuestions.map((question) => question.id)) const unknownIds = questionIds.filter((id) => !existingIds.has(id)) if (unknownIds.length > 0) { @@ -345,18 +348,30 @@ export async function POST( return NextResponse.json({ error: 'Protected clarification history is unavailable.' }, { status: 409 }) } const sourceById = new Map(existingQuestions.map((question) => [question.id, question])) - if ([...sourceById.values()].some((question) => !question.sourcePlanArtifactId || !question.sourcePlanVersion)) { + const requestedQuestions = questionIds.map((id) => sourceById.get(id)!) + if (requestedQuestions.some((question) => + question.status !== 'open' + || question.answerReferenceId !== null + || question.questionEntryId !== `clarification_question:${question.id}` + || !question.sourcePlanArtifactId + || !question.sourcePlanVersion)) { return NextResponse.json({ error: 'Clarification source is unavailable.' }, { status: 409 }) } - const appended = [] - for (const answer of answers) { + const currentSource = requestedQuestions[0] + if (existingQuestions.some((question) => + question.status === 'open' + && (question.sourcePlanArtifactId !== currentSource.sourcePlanArtifactId + || question.sourcePlanVersion !== currentSource.sourcePlanVersion))) { + return NextResponse.json({ error: 'Clarification source is unavailable.' }, { status: 409 }) + } + const appended = await appendArchitectClarificationAnswers(answers.map((answer) => { const source = sourceById.get(answer.id)! - appended.push(await appendArchitectClarificationAnswer({ + return { answer: answer.answer, digestKey: storage.digestKey, digestKeyId: storage.digestKeyId, questionId: answer.id, sessionCredential: credential, sourcePlanArtifactId: source.sourcePlanArtifactId!, sourcePlanVersion: String(source.sourcePlanVersion), taskId, - })) - } + } + })) const updatedQuestions = await db .select({ id: taskQuestions.id, @@ -378,16 +393,18 @@ export async function POST( } const { updatedQuestions, allAnswered } = result - await publishTaskEvent(taskId, 'questions:answered', { - answeredCount: updatedQuestions.length, - allAnswered, - }) - if (allAnswered) { await redis.lpush('forge:answers', JSON.stringify({ taskId })) console.info('[POST /api/tasks/:id/questions] All questions answered; enqueued re-plan', { taskId }) } + await publishTaskEvent(taskId, 'questions:answered', { + answeredCount: updatedQuestions.length, + allAnswered, + }).catch(() => { + console.warn('[POST /api/tasks/:id/questions] Answer progress event unavailable') + }) + console.info('[POST /api/tasks/:id/questions] Recorded answers', { taskId, count: updatedQuestions.length, diff --git a/web/lib/mcps/history-reader.ts b/web/lib/mcps/history-reader.ts index f6e12e20..d0353ba8 100644 --- a/web/lib/mcps/history-reader.ts +++ b/web/lib/mcps/history-reader.ts @@ -72,8 +72,7 @@ export async function readArchitectPlanHistory(input: { } } -/** Dormant B2A writer: callers must opt in explicitly during the route cutover. */ -export async function appendArchitectClarificationAnswer(input: { +export type ArchitectClarificationAnswerInput = { answer: string answerId?: string digestKey: Buffer @@ -83,27 +82,65 @@ export async function appendArchitectClarificationAnswer(input: { sourcePlanArtifactId: string sourcePlanVersion: string taskId: string -}): Promise<{ answerId: string; allAnswered: boolean }> { - const answerId = input.answerId ?? randomUUID() - const envelope = materializeArchitectClarificationAnswer({ - answer: input.answer, answerId, digestKey: input.digestKey, digestKeyId: input.digestKeyId, - questionId: input.questionId, sourcePlanArtifactId: input.sourcePlanArtifactId, - sourcePlanVersion: input.sourcePlanVersion, taskId: input.taskId, +} + +/** + * Appends one protected clarification form as one database transaction. + * + * Every answer envelope is validated before a connection is opened. The + * existing fixed-authority routine then revalidates each source and open + * question inside one transaction, so a later conflict rolls back earlier + * appends instead of partially saving the form. + */ +export async function appendArchitectClarificationAnswers( + inputs: readonly ArchitectClarificationAnswerInput[], +): Promise { + if (inputs.length < 1) { + throw new HistoryReaderError('invalid_evidence', 'The protected clarification append failed closed.') + } + const first = inputs[0] + const questionIds = new Set(inputs.map((input) => input.questionId)) + if (questionIds.size !== inputs.length + || inputs.some((input) => + input.sourcePlanArtifactId !== first.sourcePlanArtifactId + || input.sourcePlanVersion !== first.sourcePlanVersion)) { + throw new HistoryReaderError('invalid_evidence', 'The protected clarification append failed closed.') + } + const prepared = inputs.map((input) => { + if (input.taskId !== first.taskId + || input.sessionCredential !== first.sessionCredential + || input.digestKeyId !== first.digestKeyId + || !input.digestKey.equals(first.digestKey)) { + throw new HistoryReaderError('invalid_evidence', 'The protected clarification append failed closed.') + } + const answerId = input.answerId ?? randomUUID() + const envelope = materializeArchitectClarificationAnswer({ + answer: input.answer, answerId, digestKey: input.digestKey, digestKeyId: input.digestKeyId, + questionId: input.questionId, sourcePlanArtifactId: input.sourcePlanArtifactId, + sourcePlanVersion: input.sourcePlanVersion, taskId: input.taskId, + }) + return { answerId, envelope, input } }) - const credentialBytes = Buffer.from(input.sessionCredential, 'ascii') + const credentialBytes = Buffer.from(first.sessionCredential, 'ascii') const sql = postgres(historyReaderUrl(), { max: 1, prepare: true, onnotice: () => {}, transform: { undefined: null } }) try { - const [row] = await sql<{ answerId: string; allAnswered: boolean }[]>` - select answer_id as "answerId", all_answered as "allAnswered" - from forge.append_architect_clarification_answer_v1( - ${credentialBytes}::bytea, ${input.taskId}::uuid, ${input.questionId}::uuid, - ${input.sourcePlanArtifactId}::uuid, ${input.sourcePlanVersion}::bigint, - ${answerId}::uuid, ${envelope.answer}::text, ${envelope.contentDigest}::text, - ${envelope.digestKeyId}::text - ) - ` - if (!row || row.answerId !== answerId) throw new Error('missing answer append result') - return row + return await sql.begin(async (transaction) => { + const appended: { answerId: string; allAnswered: boolean }[] = [] + for (const { answerId, envelope, input } of prepared) { + const [row] = await transaction<{ answerId: string; allAnswered: boolean }[]>` + select answer_id as "answerId", all_answered as "allAnswered" + from forge.append_architect_clarification_answer_v1( + ${credentialBytes}::bytea, ${input.taskId}::uuid, ${input.questionId}::uuid, + ${input.sourcePlanArtifactId}::uuid, ${input.sourcePlanVersion}::bigint, + ${answerId}::uuid, ${envelope.answer}::text, ${envelope.contentDigest}::text, + ${envelope.digestKeyId}::text + ) + ` + if (!row || row.answerId !== answerId) throw new Error('missing answer append result') + appended.push(row) + } + return appended + }) } catch { throw new HistoryReaderError('invalid_evidence', 'The protected clarification append failed closed.') } finally { @@ -112,6 +149,14 @@ export async function appendArchitectClarificationAnswer(input: { } } +/** Single-answer compatibility wrapper around the atomic batch writer. */ +export async function appendArchitectClarificationAnswer( + input: ArchitectClarificationAnswerInput, +): Promise<{ answerId: string; allAnswered: boolean }> { + const [row] = await appendArchitectClarificationAnswers([input]) + return row +} + export async function appendProtectedMcpOperatorReview(input: { sessionCredential: string approvalGateId: string From 9340ccc01af671f1b02effd683a7d8f74078d267 Mon Sep 17 00:00:00 2001 From: Joncallim <64296013+Joncallim@users.noreply.github.com> Date: Fri, 31 Jul 2026 00:07:08 +0800 Subject: [PATCH 6/8] test: use a valid protected rollback conflict --- web/__tests__/epic-172-s4-postgres.test.ts | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/web/__tests__/epic-172-s4-postgres.test.ts b/web/__tests__/epic-172-s4-postgres.test.ts index 658dfb44..65c149ac 100644 --- a/web/__tests__/epic-172-s4-postgres.test.ts +++ b/web/__tests__/epic-172-s4-postgres.test.ts @@ -939,7 +939,7 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { sourcePlanVersion: '1', taskId, }] - await admin`update task_questions set status = 'legacy_unavailable' + await admin`delete from task_questions where task_id = ${taskId}::uuid and id = ${secondQuestionId}::uuid` await expect(appendArchitectClarificationAnswers(batch)).rejects.toMatchObject({ code: 'invalid_evidence', @@ -954,8 +954,13 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { where task_id = ${taskId}::uuid and status = 'answered') as "answeredCount"` expect(afterConflict).toEqual({ answerCount: 0, answeredCount: 0 }) - await admin`update task_questions set status = 'open' - where task_id = ${taskId}::uuid and id = ${secondQuestionId}::uuid` + await admin`insert into task_questions ( + id, task_id, question_entry_id, source_plan_artifact_id, source_plan_version, status + ) values ( + ${secondQuestionId}::uuid, ${taskId}::uuid, + ${`clarification_question:${secondQuestionId}`}, + ${source.artifactId}::uuid, 1, 'open' + )` await expect(appendArchitectClarificationAnswers(batch)).resolves.toEqual([ { answerId: firstAnswerId, allAnswered: false }, { answerId: secondAnswerId, allAnswered: true }, From 5f18dee092c956f84c2a6b97767e89d2dea54271 Mon Sep 17 00:00:00 2001 From: Joncallim <64296013+Joncallim@users.noreply.github.com> Date: Fri, 31 Jul 2026 00:01:15 +0800 Subject: [PATCH 7/8] fix: reuse the task event publisher --- web/__tests__/task-event-redis-config.test.ts | 58 ++++++++++++++++++- web/lib/task-event-redis.ts | 31 ++++++++-- 2 files changed, 83 insertions(+), 6 deletions(-) diff --git a/web/__tests__/task-event-redis-config.test.ts b/web/__tests__/task-event-redis-config.test.ts index 680db414..79b7316e 100644 --- a/web/__tests__/task-event-redis-config.test.ts +++ b/web/__tests__/task-event-redis-config.test.ts @@ -1,4 +1,4 @@ -import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' const names = [ 'REDIS_URL', @@ -13,6 +13,13 @@ describe('task-event Redis credential boundary', () => { }) afterEach(() => { + const globalTaskEvents = globalThis as typeof globalThis & { + forgeTaskEventPublisherRedis?: { disconnect: (reconnect?: boolean) => void } + forgeTaskEventPublisherRedisUrl?: string + } + globalTaskEvents.forgeTaskEventPublisherRedis?.disconnect(false) + delete globalTaskEvents.forgeTaskEventPublisherRedis + delete globalTaskEvents.forgeTaskEventPublisherRedisUrl for (const name of names) { const value = original[name] if (value === undefined) delete process.env[name] @@ -92,6 +99,55 @@ describe('task-event Redis credential boundary', () => { expect(taskEventRedisConfiguration('protected').dedicated).toBe(true) }) + it('reuses one dedicated publisher in production without adding listeners', async () => { + vi.stubEnv('NODE_ENV', 'production') + try { + const { taskEventPublisherRedis } = await import('@/lib/task-event-redis') + const configuration = { + dedicated: true, + publisherUrl: 'redis://event-publisher:publisher-password@localhost/14', + subscriberUrl: 'redis://event-subscriber:subscriber-password@localhost/14', + } + const first = taskEventPublisherRedis(configuration) + const second = taskEventPublisherRedis(configuration) + + expect(second).toBe(first) + expect(first.status).toBe('wait') + expect(first.listenerCount('error')).toBe(1) + } finally { + vi.unstubAllEnvs() + } + }) + + it('retires a closed publisher and supports deterministic test cleanup', async () => { + const { + resetTaskEventPublisherRedisForTests, + taskEventPublisherRedis, + } = await import('@/lib/task-event-redis') + const configuration = { + dedicated: true, + publisherUrl: 'redis://event-publisher:publisher-password@localhost/14', + subscriberUrl: 'redis://event-subscriber:subscriber-password@localhost/14', + } + const first = taskEventPublisherRedis(configuration) + first.disconnect(false) + expect(first.status).toBe('end') + + const replacement = taskEventPublisherRedis(configuration) + expect(replacement).not.toBe(first) + expect(replacement.status).toBe('wait') + + resetTaskEventPublisherRedisForTests() + expect((globalThis as typeof globalThis & { + forgeTaskEventPublisherRedis?: unknown + forgeTaskEventPublisherRedisUrl?: unknown + }).forgeTaskEventPublisherRedis).toBeUndefined() + expect((globalThis as typeof globalThis & { + forgeTaskEventPublisherRedisUrl?: unknown + }).forgeTaskEventPublisherRedisUrl).toBeUndefined() + expect(replacement.status).toBe('end') + }) + it('uses v2-only live and durable names even while shared legacy compatibility is configured', async () => { process.env.REDIS_URL = 'redis://legacy@localhost/0' const { diff --git a/web/lib/task-event-redis.ts b/web/lib/task-event-redis.ts index 7f899644..1cada5e0 100644 --- a/web/lib/task-event-redis.ts +++ b/web/lib/task-event-redis.ts @@ -121,15 +121,29 @@ function taskEventRedisPrincipal(redisUrl: string): string { const globalForTaskEvents = globalThis as unknown as { forgeTaskEventPublisherRedis?: Redis + forgeTaskEventPublisherRedisUrl?: string +} + +function retireTaskEventPublisherRedis(): void { + const client = globalForTaskEvents.forgeTaskEventPublisherRedis + delete globalForTaskEvents.forgeTaskEventPublisherRedis + delete globalForTaskEvents.forgeTaskEventPublisherRedisUrl + if (!client) return + client.removeAllListeners() + client.disconnect(false) } export function taskEventPublisherRedis(configuration: TaskEventRedisConfiguration): Redis { if (!configuration.dedicated) { return redis } - if (globalForTaskEvents.forgeTaskEventPublisherRedis) { - return globalForTaskEvents.forgeTaskEventPublisherRedis + const cached = globalForTaskEvents.forgeTaskEventPublisherRedis + if (cached + && globalForTaskEvents.forgeTaskEventPublisherRedisUrl === configuration.publisherUrl + && cached.status !== 'end') { + return cached } + retireTaskEventPublisherRedis() const client = new Redis(configuration.publisherUrl, { lazyConnect: true, maxRetriesPerRequest: 3, @@ -138,8 +152,15 @@ export function taskEventPublisherRedis(configuration: TaskEventRedisConfigurati client.on('error', () => { console.warn('[task-events] Publisher connection unavailable') }) - if (process.env.NODE_ENV !== 'production') { - globalForTaskEvents.forgeTaskEventPublisherRedis = client - } + globalForTaskEvents.forgeTaskEventPublisherRedis = client + globalForTaskEvents.forgeTaskEventPublisherRedisUrl = configuration.publisherUrl return client } + +/** Releases the process-scoped dedicated publisher between isolated tests. */ +export function resetTaskEventPublisherRedisForTests(): void { + if (process.env.NODE_ENV !== 'test') { + throw new Error('The task-event publisher test reset is unavailable.') + } + retireTaskEventPublisherRedis() +} From 7b7dc990b7c1c1e90cb627011ade259ab5bc4957 Mon Sep 17 00:00:00 2001 From: Joncallim <64296013+Joncallim@users.noreply.github.com> Date: Fri, 31 Jul 2026 00:05:40 +0800 Subject: [PATCH 8/8] fix: recover timestamp-only legacy claims --- .../core-diagnostic-output-closure.test.ts | 17 +++++ .../queue-occurrence-recovery.redis.test.ts | 69 +++++++++++++++++++ web/worker/queue.ts | 18 ++++- 3 files changed, 102 insertions(+), 2 deletions(-) diff --git a/web/__tests__/core-diagnostic-output-closure.test.ts b/web/__tests__/core-diagnostic-output-closure.test.ts index 64fb3d7a..bbdf0638 100644 --- a/web/__tests__/core-diagnostic-output-closure.test.ts +++ b/web/__tests__/core-diagnostic-output-closure.test.ts @@ -4598,6 +4598,12 @@ describe('core output source sentinel', () => { const malformedRecoveryMethod = queueSource.match( / private async removeMalformedRecoveryMember\([\s\S]*?\n private decodeRetryPromotionTransition/, )?.[0] ?? '' + const currentRecoveryScript = queueSource.match( + /const RECOVER_STUCK_JOB_SCRIPT = `([\s\S]*?)`\n\nconst RECOVER_LEGACY_JOB_SCRIPT/, + )?.[1] ?? '' + const legacyRecoveryScript = queueSource.match( + /const RECOVER_LEGACY_JOB_SCRIPT = `([\s\S]*?)`\n\nconst RECOVER_MALFORMED_JOB_SCRIPT/, + )?.[1] ?? '' expect(queueSource).toContain("failureCategory: DEAD_LETTER_FAILURE_CATEGORY") expect(queueSource).toContain('schemaVersion: QUEUE_ENVELOPE_SCHEMA_VERSION') @@ -4838,6 +4844,17 @@ describe('core output source sentinel', () => { ) expect(queueSource).toContain("redis.call('RPUSH', KEYS[1], ARGV[1])") expect(queueSource).toContain('const STUCK_RECOVERY_SCAN_LIMIT = 100') + expect(currentRecoveryScript).toContain('valid_marker(current_marker, now_ms)') + expect(currentRecoveryScript).not.toContain('legacy_marker_timestamp(current_marker, now_ms)') + expect(legacyRecoveryScript).toContain( + 'local timestamp = legacy_marker_timestamp(current_marker, now_ms)', + ) + expect(legacyRecoveryScript).not.toContain('valid_marker(current_marker, now_ms)') + expect(queueSource).toContain( + "not string.match(marker, '^[1-9][0-9]*$')", + ) + expect(queueSource).toContain('numeric_timestamp > 9007199254740991') + expect(queueSource).toContain('numeric_timestamp > now_ms') expect(jsonKeyScanSource).toContain('const MAX_JSON_CODE_UNITS = 1_000_000') expect(queueSource).toContain('-- forge:queue:recover-malformed-v1') expect(queueSource).toContain( diff --git a/web/__tests__/queue-occurrence-recovery.redis.test.ts b/web/__tests__/queue-occurrence-recovery.redis.test.ts index e28a431a..e4952813 100644 --- a/web/__tests__/queue-occurrence-recovery.redis.test.ts +++ b/web/__tests__/queue-occurrence-recovery.redis.test.ts @@ -2022,6 +2022,75 @@ describe.skipIf(!enabled)('queue occurrence and recovery real Redis proof', () = expect(await admin.get('forge:answers:malformed-recovery-receipts')).toBe('wrong-type') console.info('QUEUE_OCCURRENCE_REDIS_QUARANTINE_OK') + const legacyRecoveryCases = [ + { + claims: 'forge:tasks:claims', + create: () => queue(), + job: { taskId: TASK_ID, attempt: 41 }, + processing: 'forge:tasks:processing', + ready: 'forge:tasks', + }, + { + claims: 'forge:approvals:claims', + create: () => approvalQueue(), + job: { taskId: TASK_ID, action: 'approve' as const, attempt: 42 }, + processing: 'forge:approvals:processing', + ready: 'forge:approvals', + }, + { + claims: 'forge:answers:claims', + create: () => answersQueue(), + job: { taskId: TASK_ID, attempt: 43 }, + processing: 'forge:answers:processing', + ready: 'forge:answers', + }, + ] + for (const legacyCase of legacyRecoveryCases) { + await admin.del(...QUEUE_KEYS) + const raw = JSON.stringify(legacyCase.job) + const staleTimestamp = String((await redisTimeMs()) - 2_000) + await admin.rpush(legacyCase.processing, raw) + await admin.hset(legacyCase.claims, raw, staleTimestamp) + const recoveryQueue = legacyCase.create() + + await expect(recoveryQueue.recoverStuckJobs(1_000)).resolves.toBe(1) + const [recoveredRaw] = await admin.lrange(legacyCase.ready, 0, -1) + expect(parseOccurrence(recoveredRaw).job).toEqual(legacyCase.job) + expect(await admin.llen(legacyCase.processing)).toBe(0) + expect(await admin.hexists(legacyCase.claims, raw)).toBe(0) + await expect(recoveryQueue.recoverStuckJobs(1_000)).resolves.toBe(0) + + await admin.del(...QUEUE_KEYS) + const freshTimestamp = String(await redisTimeMs()) + await admin.rpush(legacyCase.processing, raw) + await admin.hset(legacyCase.claims, raw, freshTimestamp) + await expect(recoveryQueue.recoverStuckJobs(60_000)).resolves.toBe(0) + expect(await admin.lrange(legacyCase.processing, 0, -1)).toEqual([raw]) + expect(await admin.hget(legacyCase.claims, raw)).toBe(freshTimestamp) + expect(await admin.llen(legacyCase.ready)).toBe(0) + } + + const malformedLegacyMarkers = [ + '0', + '01', + '1.5', + 'NaN', + '9007199254740992', + String((await redisTimeMs()) + 10_000), + `${await redisTimeMs()}:11111111-1111-4111-8111-111111111111`, + ] + for (const marker of malformedLegacyMarkers) { + await admin.del(...QUEUE_KEYS) + const raw = JSON.stringify({ taskId: TASK_ID, attempt: 44 }) + await admin.rpush('forge:tasks:processing', raw) + await admin.hset('forge:tasks:claims', raw, marker) + await expect(queue().recoverStuckJobs(0)) + .rejects.toThrow('Queue legacy occurrence recovery failed') + expect(await admin.lrange('forge:tasks:processing', 0, -1)).toEqual([raw]) + expect(await admin.hget('forge:tasks:claims', raw)).toBe(marker) + expect(await admin.llen('forge:tasks')).toBe(0) + } + const markerCases = [ '0:11111111-1111-4111-8111-111111111111', '9007199254740992:11111111-1111-4111-8111-111111111111', diff --git a/web/worker/queue.ts b/web/worker/queue.ts index cf9cabc8..111589d8 100644 --- a/web/worker/queue.ts +++ b/web/worker/queue.ts @@ -219,6 +219,20 @@ local function valid_marker(marker, now_ms) end return true end +local function legacy_marker_timestamp(marker, now_ms) + if type(marker) ~= 'string' + or not string.match(marker, '^[1-9][0-9]*$') then + return nil + end + local numeric_timestamp = tonumber(marker) + if not numeric_timestamp + or numeric_timestamp < 1 + or numeric_timestamp > 9007199254740991 + or numeric_timestamp > now_ms then + return nil + end + return numeric_timestamp +end ` const LUA_TYPE_HELPER = ` @@ -671,10 +685,10 @@ end local now_ms = redis_now_ms() local current_marker = redis.call('HGET', KEYS[2], ARGV[1]) if current_marker then - if not valid_marker(current_marker, now_ms) then + local timestamp = legacy_marker_timestamp(current_marker, now_ms) + if not timestamp then error('forge_queue_claim_marker_invalid') end - local timestamp = tonumber(string.match(current_marker, '^([1-9][0-9]*):')) if (now_ms - timestamp) < tonumber(ARGV[3]) then if redis.call('LREM', KEYS[1], 1, ARGV[1]) ~= 1 then return 2