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__/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__/epic-172-s4-context.test.ts b/web/__tests__/epic-172-s4-context.test.ts index e98eca15..8dd242b6 100644 --- a/web/__tests__/epic-172-s4-context.test.ts +++ b/web/__tests__/epic-172-s4-context.test.ts @@ -278,6 +278,65 @@ 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 exactRoutineInventory = s4RoleBootstrap.match( + /const EXACT_CLARIFICATION_ROUTINES = \[([\s\S]*?)\] as const/, + )?.[1] ?? '' + + for (const table of [ + 'architect_clarification_answers', + 'architect_clarification_answer_writes', + ]) { + expect(ownedTableInventory).toContain(`'${table}'`) + } + for (const routine of [ + { + 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(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( + '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') + }) + 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/__tests__/epic-172-s4-postgres.test.ts b/web/__tests__/epic-172-s4-postgres.test.ts index dd3a0383..65c149ac 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,100 @@ 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`delete from task_questions + 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`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 }, + ]) + }) + it('serves protected Architect history through the real password session route with PostgreSQL as authority', async () => { const ownerPassword = 'route-history-password' const routeProject = randomUUID() @@ -1484,6 +1582,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/__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/__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/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 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() +} diff --git a/web/scripts/bootstrap-epic-172-s4-roles.ts b/web/scripts/bootstrap-epic-172-s4-roles.ts index 5f2f32a6..cd01d3fc 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', @@ -41,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("'", "''")}'` @@ -203,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 @@ -365,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 @@ -436,6 +522,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' @@ -454,7 +543,7 @@ async function main(): Promise { ) acl where acl.grantee = 0 and acl.privilege_type = 'EXECUTE' ) - ) <> 70 then + ) <> 73 then raise exception 'The S4 routine owner or PUBLIC boundary is incomplete' using errcode = '42501'; end if; 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