fix: preserve legacy Architect clarification storage

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(

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\$\$;/,

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', () => {

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',