diff --git a/README.md b/README.md
index 13dfc9c6..96476cac 100644
--- a/README.md
+++ b/README.md
@@ -1,3 +1,5 @@
+현재 구현: [영속 근거 작업·비동기 worker](docs/changes/durable-evidence-worker.md). 신규57/새전체9,572개(Python8,752+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0·독립 최종 검토 C0I0M0를 통과했습니다. SQL 송신 의도·중단 후 무재호출·공유 예약 예산·현재 권한/원문 검증을 연결했습니다. 이번 실제 AI 호출0회이며 서비스 요청·브리핑 변환·개발 배포는 후속입니다.
+
 현재 구현: [정확한 근거 필드·불확실 상태 지침](docs/changes/guarded-question-reference-rules.md). 신규47/관련532·scoped정적검사·독립 코드 검토를 통과했습니다. 실제1요청에서 모델 완료/명제3개 SDK수용·JSON 재검증을 확인했으며, 필수3명제는 AI 개발판정에서 보존을 확인했으며, 전체9,515개·정적 검사/빌드/OpenAPI도 통과했습니다. 일반 품질·앱 인수와 구분합니다.
 
 최신 실제 관측: [질문별 근거 스트림 인수](docs/research/guarded-question-acceptance.md). 요청1회/HTTP200·모델 완료·최종 사용량2,387/2,569를 확인했지만 주체/서술어 참조 누락과 불확실 상태 불일치로 SDK가 거부했습니다. 재호출/자동 보정 없이 원래 결과를 보존하며 의미·앱 인수는 남습니다.
diff --git a/apps/worker/src/issuedigest_worker/evidence_worker.py b/apps/worker/src/issuedigest_worker/evidence_worker.py
new file mode 100644
index 00000000..d96554c3
--- /dev/null
+++ b/apps/worker/src/issuedigest_worker/evidence_worker.py
@@ -0,0 +1,151 @@
+"""Explicit private durable evidence workflow; no import-time activation."""
+
+import asyncio
+from collections.abc import Callable
+from hashlib import sha256
+from typing import Any
+from uuid import UUID
+
+from dbos import DBOS
+from issuedigest_agents.guarded_question_evidence_v2 import (
+    GUARDED_QUESTION_V2_INSTRUCTIONS,
+    GuardedQuestionEvidenceGenerationV2,
+    _preflight,
+    run_guarded_question_evidence_v2,
+    validate_guarded_question_evidence_v2,
+)
+from issuedigest_agents.guarded_responses_stream import (
+    GuardedResponsesCancelled,
+    GuardedResponsesError,
+)
+from issuedigest_storage.evidence_jobs import EvidenceJobRepository
+from issuedigest_storage.research import ResearchForbidden
+from openai import AsyncOpenAI
+
+Ownership = Callable[[UUID], bool]
+
+
+def _owned(job_id: UUID, check_ownership: Ownership) -> None:
+    if check_ownership(job_id) is not True:
+        raise ResearchForbidden("Evidence worker ownership unavailable")
+
+
+def read_completed_evidence_job(
+    repository: EvidenceJobRepository, job_id: UUID, *, check_ownership: Ownership
+) -> GuardedQuestionEvidenceGenerationV2 | None:
+    """Deep current-source receipt read; SQL's opaque payload is never enough."""
+    _owned(job_id, check_ownership)
+    job, source = repository.prepare(job_id)
+    if job.status != "completed" or job.receipt is None:
+        return None
+    receipt = validate_guarded_question_evidence_v2(source, job.question, job.receipt)
+    if (
+        receipt.prompt_sha256 != job.prompt_sha256
+        or receipt.instructions_sha256 != job.instructions_sha256
+        or receipt.usage.model_dump() != job.known_usage
+        or job.usage_complete is not True
+        or job.stream_closed is not True
+    ):
+        raise ValueError("Evidence job receipt changed")
+    _owned(job_id, check_ownership)
+    return receipt
+
+
+async def execute_evidence_job(
+    repository: EvidenceJobRepository,
+    job_id: UUID,
+    *,
+    client: AsyncOpenAI,
+    check_ownership: Ownership,
+) -> GuardedQuestionEvidenceGenerationV2 | None:
+    """One native request at most, with a separately committed SQL intent."""
+    _owned(job_id, check_ownership)
+    job, source = repository.prepare(job_id)
+    if job.status == "completed":
+        return read_completed_evidence_job(repository, job_id, check_ownership=check_ownership)
+    if job.status != "pending":
+        repository.recover(job_id)
+        return None
+    try:
+        prompt, _ = _preflight(source, job.question)
+    except Exception:
+        repository.finish(job_id, reason="input")
+        raise ValueError("Evidence job input unavailable") from None
+    _owned(job_id, check_ownership)
+    if not repository.claim(
+        job_id,
+        prompt_sha256=sha256(prompt.encode()).hexdigest(),
+        instructions_sha256=sha256(GUARDED_QUESTION_V2_INSTRUCTIONS.encode()).hexdigest(),
+    ):
+        return None
+    try:
+        result = await run_guarded_question_evidence_v2(
+            client=client, model=job.model, source=source, question=job.question
+        )
+    except (GuardedResponsesError, GuardedResponsesCancelled) as error:
+        repository.finish(
+            job_id,
+            known_usage=error.known_usage.model_dump() if error.known_usage is not None else None,
+            usage_complete=error.usage_complete,
+            stream_closed=error.stream_closed,
+            reason="cancelled" if isinstance(error, asyncio.CancelledError) else error.reason,
+        )
+        raise
+    except asyncio.CancelledError:
+        repository.finish(job_id, reason="cancelled")
+        raise
+    except Exception:
+        repository.finish(job_id, reason="unknown")
+        raise ValueError("Evidence job outcome unavailable") from None
+    # Persist observations independently if current authority blocks acceptance.
+    try:
+        _owned(job_id, check_ownership)
+        current, current_source = repository.prepare(job_id)
+        validate_guarded_question_evidence_v2(current_source, current.question, result)
+        saved = repository.finish(
+            job_id,
+            receipt=result.model_dump(mode="json"),
+            known_usage=result.usage.model_dump(),
+            usage_complete=True,
+            stream_closed=True,
+        )
+    except Exception:
+        # A successful SQL commit followed by a lost acknowledgement must win.
+        repository.finish(
+            job_id,
+            known_usage=result.usage.model_dump(),
+            usage_complete=True,
+            stream_closed=True,
+            reason="input",
+        )
+        return read_completed_evidence_job(repository, job_id, check_ownership=check_ownership)
+    if saved.status != "completed":
+        return None
+    return read_completed_evidence_job(repository, job_id, check_ownership=check_ownership)
+
+
+def register_evidence_workflow(
+    repository: EvidenceJobRepository, *, client: AsyncOpenAI, check_ownership: Ownership
+) -> Callable[..., Any]:
+    """Register only; caller owns DBOS configuration, launch and workflow starts."""
+
+    @DBOS.step(name="issuedigest_guarded_evidence_call_v1", retries_allowed=False)
+    async def call(job_id: str) -> None:
+        await execute_evidence_job(
+            repository, UUID(job_id), client=client, check_ownership=check_ownership
+        )
+
+    @DBOS.workflow(name="issuedigest_guarded_evidence_job_v1")
+    async def workflow(job_id: str) -> None:
+        parsed = UUID(job_id)
+        if (
+            str(parsed) != job_id
+            or parsed.version != 4
+            or DBOS.workflow_id != "guarded-evidence-" + job_id
+        ):
+            raise ValueError("Invalid evidence workflow identity")
+        _owned(parsed, check_ownership)
+        await call(job_id)
+        read_completed_evidence_job(repository, parsed, check_ownership=check_ownership)
+
+    return workflow
diff --git a/apps/worker/tests/test_evidence_worker.py b/apps/worker/tests/test_evidence_worker.py
new file mode 100644
index 00000000..faec9255
--- /dev/null
+++ b/apps/worker/tests/test_evidence_worker.py
@@ -0,0 +1,544 @@
+"""Borrowed native transport and real SQL call-intent integration."""
+
+import asyncio
+import importlib
+import sys
+from decimal import Decimal
+from pathlib import Path
+from uuid import uuid4
+
+import httpx2 as httpx
+import pytest
+from openai import AsyncOpenAI
+from sqlalchemy import text
+
+
+@pytest.fixture(scope="module", autouse=True)
+def restore_introduced_modules():
+    original_paths = sys.path.copy()
+    sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "packages/agents/tests"))
+    sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "packages/storage/tests"))
+    packages = [
+        importlib.import_module(name) for name in ("issuedigest_agents", "issuedigest_worker")
+    ]
+    existing = set(sys.modules)
+    attributes = [vars(package).copy() for package in packages]
+    yield
+    for name in set(sys.modules) - existing:
+        if name.startswith(("issuedigest_agents.", "issuedigest_worker.evidence_worker")):
+            sys.modules.pop(name, None)
+    sys.path[:] = original_paths
+    for package, original in zip(packages, attributes, strict=True):
+        for name in set(vars(package)) - original.keys():
+            delattr(package, name)
+        for name, value in original.items():
+            setattr(package, name, value)
+
+
+def native_body():
+    import json
+
+    from test_guarded_responses_stream import Body, events
+
+    data = dict(
+        propositions=[], coverage=dict(linked_proposition_ids=[], unresolved_reason="insufficient")
+    )
+    stream = events(json.dumps(data))
+    for row in stream:
+        for item in [row.get("item"), *row.get("response", {}).get("output", [])]:
+            if item is not None:
+                item["name"] = "fill_question_evidence_v2"
+    return Body(stream)
+
+
+async def exercise(jobs, job, *, mutate=None, body=None, ownership=None):
+    requests = []
+
+    def serve(request):
+        requests.append(request)
+        if mutate:
+            mutate()
+        return httpx.Response(
+            200, headers={"content-type": "text/event-stream"}, stream=body or native_body()
+        )
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(serve)) as http:
+        client = AsyncOpenAI(api_key="synthetic", http_client=http, max_retries=0)
+        worker = importlib.import_module("issuedigest_worker.evidence_worker")
+        result = await worker.execute_evidence_job(
+            jobs, job.job_id, client=client, check_ownership=ownership or (lambda _: True)
+        )
+        assert not client.is_closed()
+        return result, requests
+
+
+def submit(jobs, engine):
+    token, workspace, snapshot = seed(engine)
+    job = jobs.submit(
+        token,
+        workspace,
+        uuid4(),
+        snapshot,
+        "합성 질문?",
+        model="gpt-6.1-sol",
+        reservation_usd=Decimal(".50"),
+    )
+    return job, token, workspace
+
+
+def test_native_complete_and_replay(jobs, engine):
+    job, token, workspace = submit(jobs, engine)
+    result, requests = asyncio.run(exercise(jobs, job))
+    assert result.status == "unreviewed"
+    assert result.evidence.status == "unreviewed"
+    assert len(requests) == 1
+    replay, calls = asyncio.run(exercise(jobs, job))
+    assert replay == result and calls == []
+    assert jobs.get(token, workspace, job.job_id).status == "completed"
+
+
+def test_committed_intent_never_reissues(jobs, engine):
+    job, _, _ = submit(jobs, engine)
+    jobs.claim(job.job_id, prompt_sha256="a" * 64, instructions_sha256="b" * 64)
+    result, requests = asyncio.run(exercise(jobs, job))
+    assert result is None and requests == []
+    assert jobs.recover(job.job_id).status == "indeterminate"
+
+
+def test_cancel_during_call_retains_usage(jobs, engine):
+    job, token, workspace = submit(jobs, engine)
+    result, requests = asyncio.run(
+        exercise(jobs, job, mutate=lambda: jobs.cancel(token, workspace, job.job_id))
+    )
+    assert result is None and len(requests) == 1
+    saved = jobs.get(token, workspace, job.job_id)
+    assert saved.status == "cancelled" and saved.receipt is None
+    assert saved.known_usage == dict(input_tokens=10, output_tokens=5, requests=1)
+
+
+def test_cached_receipt_rechecks_ownership(jobs, engine):
+    job, _, _ = submit(jobs, engine)
+    asyncio.run(exercise(jobs, job))
+    with pytest.raises(importlib.import_module("issuedigest_storage.research").ResearchForbidden):
+        asyncio.run(exercise(jobs, job, ownership=lambda _: False))
+
+
+def test_actual_dbos_registration_and_checkpoint(jobs, engine, database_url):
+    from dbos import DBOS, SetWorkflowID
+
+    job, _, _ = submit(jobs, engine)
+    requests = []
+    owned = []
+
+    def ownership(job_id):
+        owned.append(job_id)
+        return True
+
+    async def run():
+        def serve(request):
+            requests.append(request)
+            return httpx.Response(
+                200, headers={"content-type": "text/event-stream"}, stream=native_body()
+            )
+
+        async with httpx.AsyncClient(transport=httpx.MockTransport(serve)) as http:
+            client = AsyncOpenAI(api_key="synthetic", http_client=http, max_retries=0)
+            DBOS.migrate(database_url, application_role=engine.url.username)
+            DBOS(
+                config={
+                    "name": "durable-evidence-synthetic",
+                    "system_database_url": database_url,
+                    "application_version": "guarded-evidence-v1",
+                    "run_admin_server": False,
+                }
+            )
+            try:
+                workflow = importlib.import_module(
+                    "issuedigest_worker.evidence_worker"
+                ).register_evidence_workflow(jobs, client=client, check_ownership=ownership)
+                DBOS.launch()
+                identifier = "guarded-evidence-" + str(job.job_id)
+                with SetWorkflowID(identifier):
+                    handle = await DBOS.start_workflow_async(workflow, str(job.job_id))
+                assert await handle.get_result() is None
+                with SetWorkflowID(identifier):
+                    replay = await DBOS.start_workflow_async(workflow, str(job.job_id))
+                assert await replay.get_result() is None
+                assert len(requests) == 1 and len(owned) >= 4
+                assert jobs.prepare(job.job_id)[0].status == "completed"
+            finally:
+                DBOS.destroy(destroy_registry=True)
+
+    asyncio.run(run())
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("known_usage", {"input_tokens": True, "output_tokens": 5, "requests": 1}),
+        ("usage_complete", False),
+        ("receipt_sha256", "f" * 64),
+    ],
+)
+def test_corrupt_completion_rejected(jobs, engine, field, value):
+    from issuedigest_storage.evidence_jobs import evidence_jobs
+    from sqlalchemy import update
+
+    job, _, _ = submit(jobs, engine)
+    asyncio.run(exercise(jobs, job))
+    with engine.begin() as conn:
+        conn.execute(
+            update(evidence_jobs)
+            .where(evidence_jobs.c.job_id == job.job_id)
+            .values(**{field: value})
+        )
+    from issuedigest_storage.repository import StorageCorruption
+
+    with pytest.raises((ValueError, StorageCorruption)):
+        asyncio.run(exercise(jobs, job))
+
+
+@pytest.mark.parametrize("mutation", ["revoke", "title", "delete", "source"])
+def test_mutation_during_native_call_blocks_result_preserves_usage(jobs, engine, mutation):
+    from issuedigest_storage.repository import StorageCorruption
+    from issuedigest_storage.research import ResearchConflict, ResearchForbidden, ResearchRepository
+    from issuedigest_storage.schema import evidence_snapshots
+    from sqlalchemy import delete
+
+    job, _, workspace = submit(jobs, engine)
+    _, source = jobs.prepare(job.job_id)
+
+    def mutate():
+        if mutation == "revoke":
+            with engine.begin() as conn:
+                conn.execute(text("UPDATE auth_workspace_memberships SET active=false"))
+        elif mutation == "source":
+            with engine.begin() as conn:
+                conn.execute(
+                    text(
+                        "UPDATE parsed_representations SET payload = "
+                        "jsonb_set(payload, '{text}', '\"changed\"'::jsonb)"
+                    )
+                )
+        elif mutation == "title":
+            ResearchRepository(engine).upsert_presentation(
+                source[2].model_copy(update={"title": "changed"})
+            )
+        else:
+            with engine.begin() as conn:
+                conn.execute(
+                    text("DELETE FROM outbox WHERE snapshot_id=:sid"), {"sid": job.snapshot_id}
+                )
+                conn.execute(
+                    text("DELETE FROM snapshot_citations WHERE snapshot_id=:sid"),
+                    {"sid": job.snapshot_id},
+                )
+                conn.execute(
+                    delete(evidence_snapshots).where(
+                        evidence_snapshots.c.snapshot_id == job.snapshot_id
+                    )
+                )
+
+    with pytest.raises((ResearchForbidden, ResearchConflict, StorageCorruption)):
+        asyncio.run(exercise(jobs, job, mutate=mutate))
+    with engine.connect() as conn:
+        saved = conn.execute(text("SELECT status,receipt,known_usage FROM evidence_jobs")).one()
+        assert saved.status == "failed" and saved.receipt is None
+        assert saved.known_usage == dict(input_tokens=10, output_tokens=5, requests=1)
+
+
+def test_completion_commit_ack_loss_keeps_winning_result(jobs, engine, monkeypatch):
+    job, _, _ = submit(jobs, engine)
+    finish = jobs.finish
+    lost = False
+
+    def commit_then_raise(*args, **kwargs):
+        nonlocal lost
+        result = finish(*args, **kwargs)
+        if kwargs.get("receipt") is not None and not lost:
+            lost = True
+            raise RuntimeError("synthetic lost commit acknowledgement")
+        return result
+
+    monkeypatch.setattr(jobs, "finish", commit_then_raise)
+    result, requests = asyncio.run(exercise(jobs, job))
+    assert result.status == "unreviewed" and len(requests) == 1
+    assert jobs.prepare(job.job_id)[0].status == "completed"
+
+
+def test_native_cancellation_persists_observations(jobs, engine):
+    from test_guarded_responses_stream import Body
+
+    job, _, _ = submit(jobs, engine)
+
+    async def run():
+        body = Body([], wait=True)
+        task = asyncio.create_task(exercise(jobs, job, body=body))
+        await body.blocked.wait()
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        return body
+
+    body = asyncio.run(run())
+    saved = jobs.prepare(job.job_id)[0]
+    assert saved.status == "cancelled" and saved.known_usage is None
+    assert saved.stream_closed is True and body.closed
+
+
+@pytest.fixture
+def jobs(engine):
+    from issuedigest_storage.evidence_jobs import EvidenceJobRepository
+
+    with engine.begin() as conn:
+        conn.execute(text("TRUNCATE evidence_jobs"))
+    yield EvidenceJobRepository(engine)
+    with engine.begin() as conn:
+        conn.execute(text("TRUNCATE evidence_jobs"))
+
+
+def seed(engine):
+    from test_evidence_jobs import seed as make_seed
+
+    return make_seed(engine)
+
+
+def test_worker_exists():
+    name = "issuedigest_worker.evidence_worker"
+    assert importlib.util.find_spec(name) is not None, "Missing durable evidence worker"
+    assert importlib.import_module(name).execute_evidence_job
+
+
+@pytest.mark.parametrize(
+    "mode,reason,known,complete",
+    [
+        ("invalid", "output", True, True),
+        ("overcap", "usage", True, True),
+        ("transport", "transport", False, False),
+        ("whitespace", "whitespace", False, False),
+    ],
+)
+def test_native_terminal_observations(jobs, engine, mode, reason, known, complete):
+    from issuedigest_agents.guarded_responses_stream import GuardedResponsesError
+    from test_guarded_responses_stream import Body, events
+
+    job, _, _ = submit(jobs, engine)
+    body = native_body()
+    if mode == "invalid":
+        body = native_body()
+        # A terminal syntactically valid call with an invalid schema cannot become evidence.
+        for row in body.rows:
+            if "arguments" in row:
+                row["arguments"] = "{}"
+            if row.get("item", {}).get("status") == "completed":
+                row["item"]["arguments"] = "{}"
+            for item in row.get("response", {}).get("output", []):
+                item["arguments"] = "{}"
+            if "delta" in row:
+                row["delta"] = "{}"
+    elif mode == "overcap":
+        body.rows[-1]["response"]["usage"] = dict(
+            input_tokens=20001, output_tokens=5, total_tokens=20006
+        )
+    elif mode == "transport":
+        body = Body([], read_error=True)
+    elif mode == "whitespace":
+        body = Body(events(" " * 100000))
+        for row in body.rows:
+            if row.get("item"):
+                row["item"]["name"] = "fill_question_evidence_v2"
+            for item in row.get("response", {}).get("output", []):
+                item["name"] = "fill_question_evidence_v2"
+    with pytest.raises(GuardedResponsesError) as got:
+        asyncio.run(exercise(jobs, job, body=body))
+    saved = jobs.prepare(job.job_id)[0]
+    assert saved.status == "failed" and saved.receipt is None
+    assert saved.reason == got.value.reason == reason
+    assert (saved.known_usage is not None) == known
+    assert saved.usage_complete == complete and saved.stream_closed is True
+    if mode == "overcap":
+        assert saved.known_usage["input_tokens"] == 20001
+
+
+def test_imports_do_no_database_network_or_model_work():
+    import subprocess
+
+    script = """
+import socket, sqlalchemy, openai
+from dbos import DBOS
+def forbidden(*args,**kwargs):
+    raise AssertionError('Import performed work')
+socket.socket.connect=forbidden
+sqlalchemy.create_engine=forbidden
+openai.AsyncOpenAI.__init__=forbidden
+DBOS.__init__=forbidden
+import issuedigest_storage.evidence_jobs
+import issuedigest_worker.evidence_worker
+"""
+    result = subprocess.run([sys.executable, "-c", script], capture_output=True, text=True)
+    assert result.returncode == 0, result.stderr
+
+
+def test_receipt_semantics_rechecked_even_with_recomputed_hash(jobs, engine):
+    from issuedigest_agents.guarded_responses_stream import GuardedResponsesError
+    from issuedigest_storage.evidence_jobs import digest, evidence_jobs
+    from sqlalchemy import select, update
+
+    job, _, _ = submit(jobs, engine)
+    asyncio.run(exercise(jobs, job))
+    with engine.begin() as conn:
+        receipt = conn.scalar(
+            select(evidence_jobs.c.receipt).where(evidence_jobs.c.job_id == job.job_id)
+        )
+        receipt["evidence"]["question_sha256"] = "f" * 64
+        conn.execute(
+            update(evidence_jobs)
+            .where(evidence_jobs.c.job_id == job.job_id)
+            .values(receipt=receipt, receipt_sha256=digest(receipt))
+        )
+    with pytest.raises(GuardedResponsesError):
+        asyncio.run(exercise(jobs, job))
+
+
+def test_actual_dbos_cached_step_rechecks_current_authority(
+    jobs, engine, database_url, monkeypatch
+):
+    from dbos import DBOS, SetWorkflowID
+    from issuedigest_storage.research import ResearchForbidden
+
+    worker = importlib.import_module("issuedigest_worker.evidence_worker")
+    job, _, _ = submit(jobs, engine)
+    requests = []
+    reads = 0
+    original = worker.read_completed_evidence_job
+
+    def crash_after_checkpoint(*args, **kwargs):
+        nonlocal reads
+        reads += 1
+        if reads == 2:
+            raise asyncio.CancelledError("synthetic workflow interruption after committed step")
+        return original(*args, **kwargs)
+
+    monkeypatch.setattr(worker, "read_completed_evidence_job", crash_after_checkpoint)
+
+    async def run():
+        def serve(request):
+            requests.append(request)
+            return httpx.Response(
+                200, headers={"content-type": "text/event-stream"}, stream=native_body()
+            )
+
+        async with httpx.AsyncClient(transport=httpx.MockTransport(serve)) as http:
+            client = AsyncOpenAI(api_key="synthetic", http_client=http, max_retries=0)
+            DBOS.migrate(database_url, application_role=engine.url.username)
+            DBOS(
+                config={
+                    "name": "durable-evidence-synthetic",
+                    "system_database_url": database_url,
+                    "application_version": "guarded-evidence-v1",
+                    "run_admin_server": False,
+                }
+            )
+            try:
+                workflow = worker.register_evidence_workflow(
+                    jobs, client=client, check_ownership=lambda _: True
+                )
+                DBOS.launch()
+                identifier = "guarded-evidence-" + str(job.job_id)
+                with SetWorkflowID(identifier):
+                    handle = await DBOS.start_workflow_async(workflow, str(job.job_id))
+                with pytest.raises(asyncio.CancelledError):
+                    await handle.get_result()
+                steps = await DBOS.list_workflow_steps_async(identifier)
+                assert (
+                    len(steps) == 1
+                    and steps[0]["function_name"] == "issuedigest_guarded_evidence_call_v1"
+                )
+                assert steps[0]["output"] is None
+                monkeypatch.setattr(worker, "read_completed_evidence_job", original)
+                with engine.begin() as conn:
+                    conn.execute(text("UPDATE auth_workspace_memberships SET active=false"))
+                replay = await DBOS.resume_workflow_async(identifier)
+                with pytest.raises(ResearchForbidden):
+                    await replay.get_result()
+                assert len(requests) == 1
+            finally:
+                DBOS.destroy(destroy_registry=True)
+
+    asyncio.run(run())
+
+
+def test_lost_claim_acknowledgement_blocks_all_native_calls(jobs, engine, monkeypatch):
+    from issuedigest_storage.evidence_jobs import EvidenceJobRepository
+
+    job, _, _ = submit(jobs, engine)
+    claim = jobs.claim
+
+    def commit_then_raise(*args, **kwargs):
+        assert claim(*args, **kwargs)
+        raise RuntimeError("synthetic claim acknowledgement loss")
+
+    monkeypatch.setattr(jobs, "claim", commit_then_raise)
+    with pytest.raises(RuntimeError, match="acknowledgement loss"):
+        asyncio.run(exercise(jobs, job))
+    restarted = EvidenceJobRepository(engine)
+    result, requests = asyncio.run(exercise(restarted, job))
+    assert result is None and requests == []
+    assert restarted.prepare(job.job_id)[0].status == "indeterminate"
+
+
+def test_observation_storage_failure_surfaces_without_reissuing(jobs, engine, monkeypatch):
+    from issuedigest_storage.evidence_jobs import EvidenceJobRepository
+    from issuedigest_storage.repository import StorageError
+
+    job, _, _ = submit(jobs, engine)
+    body = native_body()
+
+    def unavailable(*args, **kwargs):
+        raise StorageError("Synthetic observation storage unavailable")
+
+    monkeypatch.setattr(jobs, "finish", unavailable)
+    with pytest.raises(StorageError):
+        asyncio.run(exercise(jobs, job, body=body))
+    assert body.closed
+    restarted = EvidenceJobRepository(engine)
+    assert restarted.prepare(job.job_id)[0].status == "running"
+    result, requests = asyncio.run(exercise(restarted, job))
+    assert result is None and requests == []
+    assert restarted.prepare(job.job_id)[0].status == "indeterminate"
+
+
+def test_preflight_failure_is_terminal_without_a_native_call(jobs, engine):
+    from hashlib import sha256
+
+    from issuedigest_domain import EvidenceBundle
+    from issuedigest_storage import EvidenceRepository
+
+    from conftest import bundle_data
+
+    token, workspace, _ = seed(engine)
+    data = bundle_data()
+    data["document"]["workspace_id"] = str(workspace)
+    data["representation"]["text"] = "x" * 100000
+    data["representation"]["text_sha256"] = sha256(
+        data["representation"]["text"].encode()
+    ).hexdigest()
+    data["citations"] = data["citations"][:1]
+    data["citations"][0]["selector"] = dict(
+        kind="text", start=0, end=100000, quote=data["representation"]["text"]
+    )
+    snapshot = uuid4()
+    EvidenceRepository(engine).save(snapshot, EvidenceBundle.model_validate(data))
+    job = jobs.submit(
+        token,
+        workspace,
+        uuid4(),
+        snapshot,
+        "q",
+        model="gpt-6.1-sol",
+        reservation_usd=Decimal(".25"),
+    )
+    with pytest.raises(ValueError):
+        asyncio.run(exercise(jobs, job))
+    saved = jobs.prepare(job.job_id)[0]
+    assert saved.status == "failed" and saved.known_usage is None and saved.reason == "input"
diff --git a/docs/README.md b/docs/README.md
index 7e452b62..9ca2b5b7 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -1,3 +1,5 @@
+현재 구현: [영속 근거 작업·비동기 worker](changes/durable-evidence-worker.md). 신규57/새전체9,572개(Python8,752+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0·독립 최종 검토 C0I0M0를 통과했습니다. SQL 송신 의도·중단 후 무재호출·공유 예약 예산·현재 권한/원문 검증을 연결했습니다. 이번 실제 AI 호출0회이며 서비스 요청·브리핑 변환·개발 배포는 후속입니다.
+
 현재 구현: [정확한 근거 필드·불확실 상태 지침](changes/guarded-question-reference-rules.md). 신규47/관련532·scoped정적검사·독립 코드 검토를 통과했습니다. 실제1요청에서 모델 완료/명제3개 SDK수용·JSON 재검증을 확인했으며, 필수3명제는 AI 개발판정에서 보존을 확인했으며, 전체9,515개·정적 검사/빌드/OpenAPI도 통과했습니다. 일반 품질·앱 인수와 구분합니다.
 
 최신 실제 관측: [질문별 근거 스트림 인수](research/guarded-question-acceptance.md). 요청1회/HTTP200·모델 완료·최종 사용량2,387/2,569를 확인했지만 주체/서술어 참조 누락과 불확실 상태 불일치로 SDK가 거부했습니다. 재호출/자동 보정 없이 원래 결과를 보존하며 의미·앱 인수는 남습니다.
diff --git a/docs/changes/durable-evidence-worker.md b/docs/changes/durable-evidence-worker.md
new file mode 100644
index 00000000..27380c1d
--- /dev/null
+++ b/docs/changes/durable-evidence-worker.md
@@ -0,0 +1,41 @@
+# 중단 가능한 근거 생성의 영속 작업·DBOS 연결
+
+2026-10-26 KST. [guarded-v2 실제 인수](../research/guarded-question-reference-acceptance.md) 후 내부 worker 연결이다. 기존 ResearchRun/DB는 gpt-5.4-mini·0.25예약·v9출력2,500에 고정돼 새 SDK의 모델/최대4,000출력을 그 행에 거짓 기록하지 않는다. 새 내부 근거 작업을 별도 영속 행과 DBOS workflow에 연결한다. 같은 월 예산 pool을 공유하며 기존 앱/API/worker는 활성 경로로 유지한다. 새 작업은 관리 UI나 임시 체크포인트가 아니라 후속 브리핑 파이프라인이 소비할 내부 unreviewed 산출물이다.
+
+## Global Constraints
+
+- Existing476 tracked non-Markdown files and all past namespaces remain byte-identical. Owned NEW files only: `packages/storage/src/issuedigest_storage/evidence_jobs.py`, `packages/storage/migrations/versions/0010_evidence_jobs.py`, `apps/worker/src/issuedigest_worker/evidence_worker.py`, `packages/storage/tests/test_evidence_jobs.py`, `apps/worker/tests/test_evidence_worker.py`. Documentation/this Task1 checkbox may change. No existing API/CLI/exports/dependency/old worker registry/default model/UI edits. No actual provider calls, server or operational DB changes, automatic migrations/worker launch on import, main merge or live activation.
+- Layering: storage cannot import agents/worker/dbos. Agents already depend on storage. Storage uses domain+SQLAlchemy/Pydantic and existing private auth/evidence/presentation helpers narrowly; worker imports storage and the unchanged guarded-v2 SDK. Opaque private job receipt is not ResearchResult/Canvas or semantic/public/access approval. Receipt status remains unreviewed.
+- New `EvidenceJobRepository(engine)` supports explicit `submit(token, workspace_id, request_id, snapshot_id, question, *, model, reservation_usd)`, `get(token, workspace_id, job_id)`, `cancel(token, workspace_id, job_id)`, internal `prepare(job_id)`, atomic `claim(job_id, *, prompt_sha256, instructions_sha256)`, `finish(...)`, `recover(job_id)`. Method argument details/DTO internal shapes can be chosen coherently by implementer and documented in report. submit is server-side API only, not a browser endpoint. UUIDs exact UUID4; exact nonblank question<=1500 preserved; model exact `gpt-6.1-sol`; mandatory server-configured reservation exact finite Decimal, >0..10.00, integral cents (no float/bool/coercion). No implicit price estimate/default; reservation is conservative admission accounting, not billed-cost proof.
+- Same current session/member rules as ResearchRepository: submit/cancel owner/editor only; get current workspace read membership, never cross-workspace/source. Existing shared validation scope, not a reclassification of historical data as personal/public. Submission uses the exact current EvidenceRepository source and SourcePresentation, stores source hash matching agents `_source` canonical SHA256 of snapshot_id/bundle/presentation (UTF-8 sorted compact JSON), and exact input identity including source/model/reservation/question. Same workspace+actor+request UUID is idempotent only for identical input, no second budget reservation. Changed inputs/source metadata with same id reject. Unique active new job per workspace (pending/running); do not change old run active semantics.
+- Shared monthly `research_budget_months` is locked transactionally: existing reserved_usd<=10.00 and reservation_count<=40 remain COMBINED limits across old runs and new jobs. New job increments by mandatory reservation and1 atomically, exactly once. Never release/reset reservation on success/failure/cancel/indeterminate. No new independent budget pool or old pricing/check/accounting changes.
+- Job states pending/running/completed/failed/cancelled/indeterminate; only atomic pending->running may authorize one native call. Committed running is dispatch uncertainty, not proof actual send. Running on recovery becomes indeterminate without another HTTP call; pending may execute; completed revalidates saved original against current authority/source and never reissues. Claim commit acknowledgement loss conservatively blocks another call even if send never happened. Concurrent duplicate workers cannot both claim. Completed SQL commit acknowledgement loss must preserve the winning committed result and never be overwritten by generic failure. Unknown outcome is never changed to success/free.
+- Store source/prompt/instructions hashes, exact model/profile, opaque receipt+canonical receipt hash and observed known_usage (nullable JSON exact input_tokens/output_tokens/requests=1 nonnegative ints, no bool), usage_complete, stream_closed and fixed sanitized reason. Unknown usage is None, not0. Preserve valid observed usage above success caps, without clipping or integer-column overflow (JSONB usage suitable); monotonically retain known counts even after cancellation/revocation, without changing original reservation or reviving access/status. Terminal observation usage_complete is not success. Receipt becomes completed only after worker's current-source SDK validation and storage identity checks; never accept partial/incomplete/error output as result. Failed/cancelled/indeterminate expose no receipt. Receipt hash provides corruption detection, not a signature.
+- Recheck write membership, source integrity/current metadata and exact input before claim and after await/on cached replay. Cancellation or revocation after call blocks result but retains known usage. Raw successful receipt may remain only as inaccessible audit data or be omitted; never return it to unauthorized/cancelled callers. Source deletion/tampering/metadata change rejects, not substitutes newer source. No filling/downgrading/rewriting output. Internal storage reads return opaque job data; public worker result helper must additionally run `validate_guarded_question_evidence_v2` against current source before returning a completed receipt. No known-usage write depends on active authority/session.
+- Use separate MetaData for new table to avoid import altering old metadata registry. Migration revision `0010_evidence_jobs`, down_revision `0009_policy_intake`; upgrade creates only new table/indexes via explicit connection; downgrade drops only it. No old schema/data changes. Table names/columns/DTO shape otherwise choose minimal design. Base migrations assumed present. New test fixture explicitly truncates new table before/after own tests (base fixture does not know this table), never edit global conftest; no pollution of old tests/migration snapshots.
+- Worker exports async `execute_evidence_job(repository, job_id, *, client, check_ownership)` using borrowed required AsyncOpenAI and job-stored model; no client creation/close, sync loop bridge, env/key reads, retries/repair, or provider defaults. Before claim rebuild exact guarded-v2 preflight/prompt/instructions hash and enforce unchanged20,000/4,000/60+1s bounds. Recheck ownership before/after I/O and before storage. SDK success validates original/current source and stores exact new receipt. Guarded error/cancellation observations persist before fixed error/termination; cancellation propagates CancelledError. Unexpected exception records sanitized unknown outcome, no invented usage/close. SQL storage retries must never wrap or reissue HTTP. Storage failure that prevents observation persistence surfaces safely and leaves conservative running/indeterminate state.
+- Explicit `register_evidence_workflow(repository, *, client, check_ownership)` registers DBOS async workflow `issuedigest_guarded_evidence_job_v1` and step `issuedigest_guarded_evidence_call_v1` with retries_allowed=False; workflow ID must equal `guarded-evidence-<UUID4 job>`. Caller configures/migrates/launches DBOS and starts workflow, not import/factory. Return registered callable; payload/checkpoint only job ID/None, not raw sources/credentials. SQL job is authoritative external-call intent even if DBOS step outcome missing: replay cannot repeat already-dispatched call. After cached step, recheck current authority/source and deep validate saved receipt. No automatic new-queue polling/old workflow consumption, no changes to historical step names/slots/model. Explicit private callable is usable by future service composition while keeping current default untouched.
+
+## Task 1: Durable job repository and registered async worker
+
+- [x] TDD the five owned files and migration/worker integration.
+
+Read test-driven-development. Use dedicated synthetic PostgreSQL URL supplied by controller (see dispatch), existing auth/evidence public fixtures, real native AsyncOpenAI/httpx2.MockTransport SSE (no provider), and actual installed DBOS for integration. RED missing module before implementation; GREEN focused tests + Ruff/format/mypy. Meaningful tests: migration upgrade/downgrade only new table and old rows unchanged; import has no DB/network/model work; cross-layer source hash agrees; session/role/workspace/source rejection; idempotency/change conflicts; shared old/new budget exact boundaries/concurrency; atomic duplicate claims; persisted intent then process/restart-style recovery ->indeterminate0additionalHTTP; completed receipt/job replay ->0additionalHTTP; commit-then-raise success remains completed; cancel/revoke/source/title mutation during model call blocks result retains usage/reservation; terminal invalid/output/whitespace/transport/cancel observations; unknown usage staysNone; over-cap observed usage persists; opaque receipt/profile/hash/type corruption rejected by deep worker read helper; completed result unreviewed/never Canvas. Actual DBOS registration/start/checkpoint replay verifies fixed workflow ID/name, one native request, and current authority checked even cached step. No existing paid runner, private fixture, server or production change. Keep new fixtures isolated; fullQA controller-owned. Commit only owned files+Task1 checkbox. Report exact RED/GREEN/static, public interfaces, limitations and self-review; no subagent spawning.
+
+## Controller gates
+
+- [x] Task/whole-branch review, full checks and isolated actual DBOS/PostgreSQL/no-provider integration evidence.
+- [x] Historical476 execution/previous evidence preservation, docs/privacy, Draft PR, Sites summary/previous snapshot, ownQA/SDD cleanup.
+- [ ] Service-side enqueue/briefing conversion/multi-source orchestration and operator reservation policy validation before development deployment. No live activation in this change.
+
+## 판단과 영향
+
+1. 기존 모델/비용/결과 계약을 바꾸지 않고 내부 근거 작업을 별도 영속 행·DBOS workflow로 만든다. 기존 복구를 보존하지만 서비스 enqueue·브리핑 변환 연결과 개발 배포는 후속이다.
+2. 송신 의도를 먼저 영속화하고 결과가 모호하면 재호출하지 않는다. 중복 비용을 막지만 실제 송신 전 중단된 작업도 indeterminate로 남을 수 있다.
+3. 월 예약 pool은 공유하고 예약 금액은 서버가 명시해야 한다. 전체 예약 상한은 유지하지만 이 금액이 실제 청구 상한이라는 보장은 아니며 운영 비용 정책 검증이 남는다.
+
+4. 기존 auth helper가 여러 workspace에서 공유하는 사용자 행도 잠그므로 새 작업의 짧은 SQL transaction을 전용 advisory namespace로 직렬화한다. AI await까지 유지하지 않으며 잠금 역전을 방지하지만 처리량 병목이 생길 수 있어 운영 부하 인수가 남는다.
+
+## 검증 결과
+
+실행abfbe26·신규57/새전체9,572개(Python8,752+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0·최종 검토 C0I0M0. 초기 I2는 완료 취소 검증/SQL 잠금 순서 수정 후 회귀7·좁은 재검토로 해소했다. [상세 범위·한계](../research/durable-evidence-worker-validation.md)를 따른다. 실제 제공자/서버/앱 활성화0이다.
diff --git a/docs/planning/implementation-plan.md b/docs/planning/implementation-plan.md
index d416ee60..eb1d1661 100644
--- a/docs/planning/implementation-plan.md
+++ b/docs/planning/implementation-plan.md
@@ -1,3 +1,7 @@
+## M3 검증 완료 · 영속 근거 작업과 복구
+
+[내부 저장·worker 계약](../changes/durable-evidence-worker.md)·[검증 범위](../research/durable-evidence-worker-validation.md). 신규57/새전체9,572개(Python8,752+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0·독립 최종 검토 C0I0M0다. 전용 합성 PostgreSQL/실제 DBOS에서 단일 송신·중단/캐시 재개·취소/권한/원문 변경·사용량 보존을 확인했다. 기존476실행/이전68증거와 앱/기본worker/main은 유지한다. 서비스 enqueue·브리핑 변환·다자료·예약 정책/부하 인수·개발 배포가 다음이다. 실제 제공자 호출0이며 청구/의미/앱 인수를 뜻하지 않는다.
+
 ## M3 진행 · 명시적 근거 필드·불확실 상태
 
 [guarded-v2](../changes/guarded-question-reference-rules.md)는 실제 빈 필드 조건을328byte로 명시하고 이전 SDK/검증기는 보존한다. [단일 실제 인수](../research/guarded-question-reference-acceptance.md)에서 명제3개의 SDK수용/원문 검증/JSON왕복을 확인했다. 필수3명제는 비맹검 AI개발판정 explicit3/다른9미평가·독립 증거 검토 C0I0M0이며 새전체9,515개(Python8,695+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0를 통과했다. 실제 worker/합산 예산/권한/복구·앱 인수는 별도로 남는다.
diff --git a/docs/planning/progress.md b/docs/planning/progress.md
index 186b3244..686ebe8d 100644
--- a/docs/planning/progress.md
+++ b/docs/planning/progress.md
@@ -1,3 +1,9 @@
+## 검증 완료 · 영속 근거 작업·비동기 worker · 2026-10-26
+
+[새 계약](../changes/durable-evidence-worker.md)·[검증 기록](../research/durable-evidence-worker-validation.md). 실행abfbe26·신규57/새전체9,572개(Python8,752+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0·독립 최종 검토 C0I0M0·미해결0다. 최초 검토의 완료 취소 응답/잠금 역전2건은 회귀7건 RED 재현 후 수정·좁은 재검토로 해소했다. 최초 중단된 전체 검사는 통과로 세지 않고 보존했다. 기존476실행/이전68증거·앱/기본worker/main은 유지하며 제공자 호출0·운영 서버/DB 변경0이다. 서비스 요청·브리핑 변환·예약 정책/부하·개발 반영은 후속이며 PR/Sites/정리는 최종 인계를 따른다.
+
+최종 인계: [Draft PR110](https://github.com/DigestWorks/IssueDigest/pull/110)·Sitesv132 public/native succeeded, 이전v131/변경7문서 정확 사본(03ec9e9)을 보존했다. Site source `fcedd616dd2871eeb18b4d7355203e7b69a1cdab`, saved `appgprj_6ac612dd42708191a7a0c8377be64d48~appgver_a43137d78bb4819199961d654929f981`, deployment `appgdep_6adec48114e8819197f3088fd6e211ce`다. HTML117/로컬대상3,401/공개텍스트355·비공개60+12항목 제외를 확인했고 사용자 탭 이동/새브라우저QA0이다. 자체QA1/전용volume1을 정리하고 전용 포트 닫힘을 확인했다. 네 판단과 영향은 명세/검증 기록에 남겼다. 이후 인계는 저장소 원본을 따르며 앱/worker/main 활성화0이다. 최종 문서 push의 원격CI는 로컬 전체 검증과 구분한다.
+
 ## 현재 구현·실제 관측 · 정확한 근거 필드·불확실 상태 · 2026-10-26
 
 [새 SDK 계약](../changes/guarded-question-reference-rules.md)·[실제 인수](../research/guarded-question-reference-acceptance.md). 신규47/관련532·scoped정적검사·태스크/전체 코드 검토 C0I0M0. 같은 원문/Q2·새328byte 지침/envelope19,907을 사전 고정해 실제1요청/HTTP200/completed·30.925초·명제3개 SDK수용/JSON저장 재검증을 확인했다. 최종usage2,444/2,781(추론포함)을 보존했고 재호출/보정0이다. 독립 AI개발판정은 필수3명제 explicit3/다른9미평가이며 증거검토 C0I0M0다. 일반/사람/holdout 품질로 확대하지 않는다. 새전체9,515개(Python8,695+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0를 통과했다. Sites/PR 최종 인계는 아래 기록을 따른다. 기존474실행/과거 결과·비용·앱/worker/main은 유지한다.
diff --git a/docs/planning/tasks.md b/docs/planning/tasks.md
index 3bc0174d..f591d926 100644
--- a/docs/planning/tasks.md
+++ b/docs/planning/tasks.md
@@ -1,3 +1,11 @@
+## 검증 완료 · 영속 근거 작업·비동기 worker · 2026-10-26
+
+- [x] [새 작업/공유 예산/권한/재생 계약](../changes/durable-evidence-worker.md) 고정.
+- [x] 별도 job migration/repository·비동기 guarded-v2·실제 DBOS 연결 TDD.
+- [x] 불명확 송신/취소/권한·source 변경·공유 예산·현재 검증 replay 검사.
+- [x] 독립 검토·전체9,572검사·기존476실행/68이전증거 보존·Draft PR110/Sitesv132·QA 정리.
+- [ ] 서비스 enqueue/브리핑 변환·다자료/운영 예약 정책·개발 배포 인수.
+
 ## 진행 · 정확한 근거 필드·불확실 상태 · 2026-10-26
 
 - [x] [새 guarded-v2 지침/영수증](../changes/guarded-question-reference-rules.md) 고정·328byte/기존 입력 envelope19,907.
diff --git a/docs/research/durable-evidence-worker-validation.md b/docs/research/durable-evidence-worker-validation.md
new file mode 100644
index 00000000..0e4f9c7d
--- /dev/null
+++ b/docs/research/durable-evidence-worker-validation.md
@@ -0,0 +1,37 @@
+# 영속 근거 작업·DBOS 검증
+
+2026-10-26 KST. [명세](../changes/durable-evidence-worker.md)의 내부 저장·실행 연결 검증이다. 실제 앱·기본 worker·main은 변경하지 않는다. 실제 모델 호출·제공자 과금/의미 품질·브리핑 인수의 증거는 아니다.
+
+## 구현 범위
+
+신규 `EvidenceJobRepository`는 정확한 원문/질문/모델·서버 지정 예약 금액을 별도 작업에 저장한다. 기존 월 예약 pool의 10USD/40회 합산 제한을 공유하며 어떤 종료 상태에서도 예약을 돌려주지 않는다. 예약 금액은 실제 청구 상한이 아니다.
+
+`execute_evidence_job`은 현재 권한·원문을 확인하고 송신 의도를 먼저 커밋한 뒤 빌린 async client로 guarded-v2를 한 번 호출한다. `register_evidence_workflow`가 별도 DBOS workflow/step을 등록하며 호출자가 DBOS 설정·시작을 맡는다. import·등록만으로 DB/모델 작업을 실행하지 않는다. 완료 결과는 unreviewed이며 브리핑/Canvas로 자동 전환하지 않는다.
+
+## 검증 방법
+
+전용 로컬 PostgreSQL·합성 원문·실제 AsyncOpenAI/모의 HTTP SSE·설치된 DBOS를 사용했다. 제공자 호출과 운영 서버/DB 변경은 0회다. 기존 실행 파일을 고정하고 신규 테이블은 해당 테스트에서만 정리한다.
+
+초기 저장소/worker 부재에 대한 실패를 확인한 뒤 구현했다. 보완 후 집중 검사 57개와 신규 파일의 Ruff/형식/mypy를 통과했다. 초기 독립 검토에서 완료 취소 응답과 잠금 역전 2건을 발견했다. 실제 실패 7건을 재현하고 수정 후 같은 검사와 집중 검사를 통과했다. 수정 재검토와 최종 전체 검토는 C0I0M0·미해결0이다. 실행abfbe26 기준 신규57/새전체9,572개(Python8,752+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0를 통과했다. 초기 중단한 전체 실행은 별도 보존하고 통과로 세지 않았다.
+
+- 저장·입력: 원문 hash 일치, 정확한 UUID/질문/모델/Decimal 예약, 현재 세션·역할·자료 범위, 멱등성, 이전/새 작업의 합산 예산과 동시 마지막 예약.
+- 송신·복구: 동시 claim, 송신 의도 후 중단, claim 응답 유실, 완료 커밋 후 응답 유실, 완료 재독해에서 추가 HTTP 없음.
+- 결과·사용량: 취소/권한 상실/원문 삭제·변조·제목 변경 뒤 결과 거부와 관측 사용량 보존, 미관측 None, 초과/큰 정수 보존, 취소 전파, 저장 결과 hash/의미 재검증.
+- DBOS: 고정 workflow ID·이름·무재시도 step, 완료 재진입, step checkpoint 뒤 중단·재개 시 캐시를 사용하면서 현재 권한을 다시 검사. 각 모델 작업의 모의 HTTP는 한 번이다.
+- 마이그레이션: 신규 테이블만 upgrade/downgrade하고 기존 snapshot 행·metadata를 보존한다.
+
+## 한계와 다음 연결
+
+서비스에서 작업을 제출하는 경로·브리핑 변환·다중 자료 orchestration·운영 예약 정책 인수와 개발 배포는 남아 있다. 기존 UI에 새 경로를 활성화하지 않는다. 합성 DBOS/HTTP 검증으로 실제 제공자의 중단·청구·의미 품질을 보장하지 않는다.
+
+송신 의도 후 응답이 없으면 재호출하지 않고 indeterminate로 둔다. 실제 송신 직전에 중단돼도 예약과 불확실 상태가 남을 수 있다. 관측 사용량 저장도 실패하면 실행 오류를 드러내며, DB 복구 후에는 미관측/불확실 상태로 처리한다. SQL receipt hash는 서명이 아니며 소비자는 현재 원문에 대한 worker의 심층 검증을 사용해야 한다.
+
+## 선택과 영향
+
+1. 기존 모델/가격/결과를 바꾸지 않고 별도 내부 작업을 만든다. 이전 복구 계약을 보존하지만 서비스 요청·브리핑·배포 연결은 후속이다.
+2. 송신 의도를 먼저 저장하고 모호한 실행은 재호출하지 않는다. 중복 비용을 막지만 실제 호출이 없던 작업도 불확실 상태로 남을 수 있다.
+3. 기존 월 예약 pool을 공유하고 서버가 예약 금액을 명시한다. 합산 제한을 유지하지만 실제 청구 상한이 아니므로 운영 비용 정책 인수가 필요하다.
+
+4. 기존 사용자 행 잠금의 역전을 막도록 새 작업 SQL transaction에 전용 advisory gate를 둔다. AI 호출 동안에는 잠금을 보유하지 않지만 짧은 DB 처리의 동시성이 줄어 부하 인수가 필요하다.
+
+최종 인계: [Draft PR110](https://github.com/DigestWorks/IssueDigest/pull/110)·계획판v132/이전v131·변경7문서(03ec9e9) 사본 보존. 자체QA1/전용volume1만 정리했다. 브라우저QA/사용자탭 이동0·이번 실제 제공자 호출0이며 앱/기본worker/main 활성화는 없다. 게시 이후 인계 기록은 저장소 원본을 따른다.
diff --git a/packages/storage/migrations/versions/0010_evidence_jobs.py b/packages/storage/migrations/versions/0010_evidence_jobs.py
new file mode 100644
index 00000000..fadd04ab
--- /dev/null
+++ b/packages/storage/migrations/versions/0010_evidence_jobs.py
@@ -0,0 +1,17 @@
+"""Isolated opaque private evidence jobs."""
+
+from alembic import op
+from issuedigest_storage.evidence_jobs import evidence_jobs
+
+revision = "0010_evidence_jobs"
+down_revision = "0009_policy_intake"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    evidence_jobs.create(op.get_bind(), checkfirst=False)
+
+
+def downgrade() -> None:
+    evidence_jobs.drop(op.get_bind(), checkfirst=False)
diff --git a/packages/storage/src/issuedigest_storage/evidence_jobs.py b/packages/storage/src/issuedigest_storage/evidence_jobs.py
new file mode 100644
index 00000000..c5d33e94
--- /dev/null
+++ b/packages/storage/src/issuedigest_storage/evidence_jobs.py
@@ -0,0 +1,479 @@
+"""Private opaque durable jobs; external call intent is committed before dispatch."""
+
+import json
+import re
+from dataclasses import dataclass
+from decimal import Decimal
+from hashlib import sha256
+from typing import Any
+from uuid import UUID, uuid4
+
+from issuedigest_domain import EvidenceBundle
+from issuedigest_domain.research_journey import SourcePresentation
+from sqlalchemy import (
+    Boolean,
+    CheckConstraint,
+    Column,
+    Index,
+    MetaData,
+    Numeric,
+    String,
+    Table,
+    UniqueConstraint,
+    func,
+    insert,
+    select,
+    update,
+)
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.dialects.postgresql import UUID as PGUUID
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.engine import Connection, Engine
+
+from .repository import EvidenceRepository, StorageCorruption, _uuid4
+from .research import (
+    ResearchBudgetExceeded,
+    ResearchConflict,
+    ResearchForbidden,
+    ResearchRepository,
+)
+from .research_schema import research_budget_months
+
+metadata = MetaData()
+evidence_jobs = Table(
+    "evidence_jobs",
+    metadata,
+    *[
+        Column(n, PGUUID(as_uuid=True), primary_key=n == "job_id", nullable=False)
+        for n in ("job_id", "workspace_id", "actor_id", "request_id", "snapshot_id")
+    ],
+    *[
+        Column(n, String, nullable=False)
+        for n in ("question", "model", "profile", "source_sha256", "input_sha256", "status")
+    ],
+    Column("reservation_usd", Numeric(6, 2), nullable=False),
+    *[
+        Column(n, String, nullable=True)
+        for n in ("prompt_sha256", "instructions_sha256", "receipt_sha256", "reason")
+    ],
+    Column("receipt", JSONB),
+    Column("known_usage", JSONB),
+    Column("usage_complete", Boolean, nullable=False, server_default="false"),
+    Column("stream_closed", Boolean, nullable=False, server_default="false"),
+    UniqueConstraint("workspace_id", "actor_id", "request_id"),
+    CheckConstraint(
+        "status IN ('pending','running','completed','failed','cancelled','indeterminate')"
+    ),
+    CheckConstraint("model = 'gpt-6.1-sol' AND profile = 'guarded-question-evidence-v2'"),
+    CheckConstraint("reservation_usd > 0 AND reservation_usd <= 10"),
+)
+Index(
+    "evidence_jobs_active",
+    evidence_jobs.c.workspace_id,
+    unique=True,
+    postgresql_where=evidence_jobs.c.status.in_(["pending", "running"]),
+)
+
+
+def digest(value: Any) -> str:
+    return sha256(
+        json.dumps(
+            value, sort_keys=True, ensure_ascii=False, separators=(",", ":"), allow_nan=False
+        ).encode()
+    ).hexdigest()
+
+
+def source_digest(source: tuple[UUID, EvidenceBundle, SourcePresentation]) -> str:
+    sid, bundle, presentation = source
+    return digest(
+        dict(
+            snapshot_id=str(sid),
+            bundle=bundle.model_dump(mode="json"),
+            presentation=presentation.model_dump(mode="json"),
+        )
+    )
+
+
+@dataclass(frozen=True)
+class EvidenceJob:
+    job_id: UUID
+    workspace_id: UUID
+    snapshot_id: UUID
+    question: str
+    model: str
+    status: str
+    source_sha256: str
+    prompt_sha256: str | None
+    instructions_sha256: str | None
+    receipt: dict[str, Any] | None
+    known_usage: dict[str, int] | None
+    usage_complete: bool
+    stream_closed: bool
+    reason: str | None
+
+
+class EvidenceJobRepository:
+    def __init__(self, engine: Engine):
+        self.engine = engine
+        self._base = ResearchRepository(engine)
+
+    @staticmethod
+    def _lock_jobs(conn: Connection) -> None:
+        # Session/member checks also lock shared auth_users across workspaces.
+        # Order every short new-job SQL transaction before those and job row locks;
+        # this transaction gate is never held across native I/O.
+        key = int.from_bytes(
+            sha256(b"issuedigest:evidence-jobs:transaction-v1").digest()[:8],
+            "big",
+            signed=True,
+        )
+        conn.scalar(select(func.pg_advisory_xact_lock(key)))
+
+    @staticmethod
+    def _row(conn: Connection, job_id: UUID) -> Any:
+        _uuid4(job_id)
+        EvidenceJobRepository._lock_jobs(conn)
+        row = (
+            conn.execute(
+                select(evidence_jobs).where(evidence_jobs.c.job_id == job_id).with_for_update()
+            )
+            .mappings()
+            .one_or_none()
+        )
+        if row is None:
+            raise ResearchConflict("Evidence job unavailable")
+        return row
+
+    @staticmethod
+    def _dto(row: Any) -> EvidenceJob:
+        receipt = row["receipt"] if row["status"] == "completed" else None
+        if row["model"] != "gpt-6.1-sol" or row["profile"] != "guarded-question-evidence-v2":
+            raise StorageCorruption("Evidence job identity changed")
+        usage = row["known_usage"]
+        if usage is not None and (
+            type(usage) is not dict
+            or set(usage) != {"input_tokens", "output_tokens", "requests"}
+            or any(type(value) is not int or value < 0 for value in usage.values())
+            or usage["requests"] != 1
+        ):
+            raise StorageCorruption("Evidence job usage changed")
+        if row["status"] == "completed" and type(receipt) is not dict:
+            raise StorageCorruption("Evidence job receipt unavailable")
+        if receipt is not None and digest(receipt) != row["receipt_sha256"]:
+            raise StorageCorruption("Evidence job receipt changed")
+        fields: dict[str, Any] = {
+            n: receipt if n == "receipt" else row[n] for n in EvidenceJob.__dataclass_fields__
+        }
+        return EvidenceJob(**fields)
+
+    @staticmethod
+    def _current(conn: Connection, row: Any) -> tuple[UUID, EvidenceBundle, SourcePresentation]:
+        if not ResearchRepository._member(conn, row["actor_id"], row["workspace_id"], write=True):
+            raise ResearchForbidden("Evidence job authority unavailable")
+        bundle = EvidenceRepository._load(conn, row["workspace_id"], row["snapshot_id"])
+        if bundle is None:
+            raise ResearchConflict("Evidence job source unavailable")
+        presentation = ResearchRepository._presentation(
+            conn, row["workspace_id"], row["snapshot_id"], bundle
+        )
+        source = row["snapshot_id"], bundle, presentation
+        if source_digest(source) != row["source_sha256"]:
+            raise StorageCorruption("Evidence job source changed")
+        identity = dict(
+            snapshot_id=str(row["snapshot_id"]),
+            source_sha256=row["source_sha256"],
+            question=row["question"],
+            model=row["model"],
+            reservation_usd=str(row["reservation_usd"]),
+        )
+        if digest(identity) != row["input_sha256"]:
+            raise StorageCorruption("Evidence job input changed")
+        return source
+
+    def submit(
+        self,
+        token: str,
+        workspace_id: UUID,
+        request_id: UUID,
+        snapshot_id: UUID,
+        question: str,
+        *,
+        model: str,
+        reservation_usd: Decimal,
+    ) -> EvidenceJob | None:
+        for value in (workspace_id, request_id, snapshot_id):
+            _uuid4(value)
+        if (
+            type(question) is not str
+            or not question.strip()
+            or len(question) > 1500
+            or type(model) is not str
+            or model != "gpt-6.1-sol"
+        ):
+            raise ValueError("Invalid evidence job input")
+        if (
+            type(reservation_usd) is not Decimal
+            or not reservation_usd.is_finite()
+            or not Decimal("0") < reservation_usd <= Decimal("10")
+            or reservation_usd != reservation_usd.quantize(Decimal(".01"))
+        ):
+            raise ValueError("Invalid evidence reservation")
+        reservation_usd = reservation_usd.quantize(Decimal(".01"))
+
+        def save(conn: Connection) -> EvidenceJob | None:
+            self._lock_jobs(conn)
+            actor = self._base._actor(conn, token, workspace_id, write=True)
+            if actor is None:
+                return None
+            bundle = EvidenceRepository._load(conn, workspace_id, snapshot_id)
+            if bundle is None:
+                raise ResearchConflict("Evidence source unavailable")
+            presentation = self._base._presentation(conn, workspace_id, snapshot_id, bundle)
+            hashed = source_digest((snapshot_id, bundle, presentation))
+            identity = digest(
+                dict(
+                    snapshot_id=str(snapshot_id),
+                    source_sha256=hashed,
+                    question=question,
+                    model=model,
+                    reservation_usd=str(reservation_usd),
+                )
+            )
+            prior = (
+                conn.execute(
+                    select(evidence_jobs).where(
+                        evidence_jobs.c.workspace_id == workspace_id,
+                        evidence_jobs.c.actor_id == actor,
+                        evidence_jobs.c.request_id == request_id,
+                    )
+                )
+                .mappings()
+                .one_or_none()
+            )
+            if prior is not None:
+                if prior["input_sha256"] != identity:
+                    raise ResearchConflict("Evidence request changed")
+                return self._dto(prior)
+            month = conn.scalar(select(func.to_char(func.clock_timestamp(), "YYYY-MM")))
+            conn.execute(
+                pg_insert(research_budget_months).values(month=month).on_conflict_do_nothing()
+            )
+            budget = (
+                conn.execute(
+                    select(research_budget_months)
+                    .where(research_budget_months.c.month == month)
+                    .with_for_update()
+                )
+                .mappings()
+                .one()
+            )
+            if budget["reservation_count"] >= 40 or budget[
+                "reserved_usd"
+            ] + reservation_usd > Decimal("10"):
+                raise ResearchBudgetExceeded("Monthly research budget reached")
+            conn.execute(
+                update(research_budget_months)
+                .where(research_budget_months.c.month == month)
+                .values(
+                    reserved_usd=budget["reserved_usd"] + reservation_usd,
+                    reservation_count=budget["reservation_count"] + 1,
+                )
+            )
+            job_id = uuid4()
+            conn.execute(
+                insert(evidence_jobs).values(
+                    job_id=job_id,
+                    workspace_id=workspace_id,
+                    actor_id=actor,
+                    request_id=request_id,
+                    snapshot_id=snapshot_id,
+                    question=question,
+                    model=model,
+                    profile="guarded-question-evidence-v2",
+                    source_sha256=hashed,
+                    input_sha256=identity,
+                    reservation_usd=reservation_usd,
+                    status="pending",
+                )
+            )
+            return self._dto(self._row(conn, job_id))
+
+        return self._base._transaction(save)
+
+    def prepare(
+        self, job_id: UUID
+    ) -> tuple[EvidenceJob, tuple[UUID, EvidenceBundle, SourcePresentation]]:
+        def read(
+            conn: Connection,
+        ) -> tuple[EvidenceJob, tuple[UUID, EvidenceBundle, SourcePresentation]]:
+            row = self._row(conn, job_id)
+            return self._dto(row), self._current(conn, row)
+
+        return self._base._transaction(read)
+
+    def get(self, token: str, workspace_id: UUID, job_id: UUID) -> EvidenceJob | None:
+        _uuid4(workspace_id)
+        _uuid4(job_id)
+
+        def read(conn: Connection) -> EvidenceJob | None:
+            self._lock_jobs(conn)
+            if self._base._actor(conn, token, workspace_id) is None:
+                return None
+            row = self._row(conn, job_id)
+            if row["workspace_id"] != workspace_id:
+                return None
+            self._current(conn, row)
+            return self._dto(row)
+
+        return self._base._transaction(read)
+
+    def cancel(self, token: str, workspace_id: UUID, job_id: UUID) -> EvidenceJob | None:
+        _uuid4(workspace_id)
+        _uuid4(job_id)
+
+        def stop(conn: Connection) -> EvidenceJob | None:
+            self._lock_jobs(conn)
+            if self._base._actor(conn, token, workspace_id, write=True) is None:
+                return None
+            row = self._row(conn, job_id)
+            if row["workspace_id"] != workspace_id:
+                return None
+            if row["status"] == "completed":
+                self._current(conn, row)
+            if row["status"] in ("pending", "running"):
+                conn.execute(
+                    update(evidence_jobs)
+                    .where(evidence_jobs.c.job_id == job_id)
+                    .values(status="cancelled", reason="cancelled", receipt=None)
+                )
+            return self._dto(self._row(conn, job_id))
+
+        return self._base._transaction(stop)
+
+    def claim(self, job_id: UUID, *, prompt_sha256: str, instructions_sha256: str) -> bool:
+        if any(
+            type(v) is not str or re.fullmatch("[0-9a-f]{64}", v) is None
+            for v in (prompt_sha256, instructions_sha256)
+        ):
+            raise ValueError("Invalid evidence hashes")
+
+        def claim_one(conn: Connection) -> bool:
+            row = self._row(conn, job_id)
+            self._current(conn, row)
+            if row["status"] != "pending":
+                return False
+            conn.execute(
+                update(evidence_jobs)
+                .where(evidence_jobs.c.job_id == job_id)
+                .values(
+                    status="running",
+                    prompt_sha256=prompt_sha256,
+                    instructions_sha256=instructions_sha256,
+                )
+            )
+            return True
+
+        return self._base._transaction(claim_one)
+
+    def recover(self, job_id: UUID) -> EvidenceJob:
+        def recover_one(conn: Connection) -> EvidenceJob:
+            row = self._row(conn, job_id)
+            if row["status"] == "running":
+                conn.execute(
+                    update(evidence_jobs)
+                    .where(evidence_jobs.c.job_id == job_id)
+                    .values(status="indeterminate", reason="unknown")
+                )
+            return self._dto(self._row(conn, job_id))
+
+        return self._base._transaction(recover_one)
+
+    def finish(
+        self,
+        job_id: UUID,
+        *,
+        receipt: dict[str, Any] | None = None,
+        known_usage: dict[str, int] | None = None,
+        usage_complete: bool = False,
+        stream_closed: bool = False,
+        reason: str = "unknown",
+    ) -> EvidenceJob:
+        if known_usage is not None and (
+            type(known_usage) is not dict
+            or set(known_usage) != {"input_tokens", "output_tokens", "requests"}
+            or any(type(v) is not int or v < 0 for v in known_usage.values())
+            or known_usage["requests"] != 1
+        ):
+            raise ValueError("Invalid observed usage")
+        if type(usage_complete) is not bool or type(stream_closed) is not bool:
+            raise ValueError("Invalid observations")
+        if reason not in (
+            "unknown",
+            "input",
+            "output",
+            "cancelled",
+            "timeout",
+            "transport",
+            "whitespace",
+            "incomplete",
+            "protocol",
+            "usage",
+            "limit",
+        ):
+            reason = "unknown"
+
+        def complete(conn: Connection) -> EvidenceJob:
+            row = self._row(conn, job_id)
+            if row["status"] == "completed":
+                return self._dto(row)
+            previous = row["known_usage"]
+            usage = known_usage
+            if previous is not None:
+                usage = {k: max(previous[k], (known_usage or previous)[k]) for k in previous}
+            values: dict[str, Any] = dict(
+                known_usage=usage,
+                usage_complete=row["usage_complete"] or usage_complete,
+                stream_closed=row["stream_closed"] or stream_closed,
+            )
+            if receipt is not None:
+                self._current(conn, row)
+                if row["status"] == "running":
+                    if (
+                        receipt.get("prompt_sha256") != row["prompt_sha256"]
+                        or receipt.get("instructions_sha256") != row["instructions_sha256"]
+                        or receipt.get("profile") != "guarded-question-evidence-v2"
+                        or receipt.get("status") != "unreviewed"
+                        or receipt.get("transport_profile") != "guarded-responses-tool-v1"
+                        or receipt.get("input_format") != "question-evidence-source-v2"
+                        or type(receipt.get("evidence")) is not dict
+                        or receipt["evidence"].get("source_sha256") != row["source_sha256"]
+                        or receipt["evidence"].get("question_sha256")
+                        != sha256(row["question"].encode()).hexdigest()
+                        or receipt.get("usage") != usage
+                        or not usage_complete
+                        or not stream_closed
+                    ):
+                        raise StorageCorruption("Evidence receipt identity changed")
+                    values.update(
+                        status="completed",
+                        receipt=receipt,
+                        receipt_sha256=digest(receipt),
+                        reason=None,
+                    )
+            elif row["status"] in ("pending", "running"):
+                values.update(
+                    status=(
+                        "indeterminate"
+                        if reason == "unknown"
+                        else "cancelled"
+                        if reason == "cancelled"
+                        else "failed"
+                    ),
+                    reason=reason,
+                )
+            conn.execute(
+                update(evidence_jobs).where(evidence_jobs.c.job_id == job_id).values(**values)
+            )
+            return self._dto(self._row(conn, job_id))
+
+        return self._base._transaction(complete)
diff --git a/packages/storage/tests/test_evidence_jobs.py b/packages/storage/tests/test_evidence_jobs.py
new file mode 100644
index 00000000..843b93ea
--- /dev/null
+++ b/packages/storage/tests/test_evidence_jobs.py
@@ -0,0 +1,643 @@
+"""Dedicated synthetic durable job integration."""
+
+import importlib
+from decimal import Decimal
+from uuid import uuid4
+
+import pytest
+from issuedigest_domain import EvidenceBundle
+from issuedigest_storage import EvidenceRepository, IdentityRepository, SessionRepository
+from sqlalchemy import text
+
+from conftest import bundle_data
+
+
+def api():
+    name = "issuedigest_storage.evidence_jobs"
+    assert importlib.util.find_spec(name) is not None, "Missing durable job repository"
+    return importlib.import_module(name)
+
+
+@pytest.fixture
+def jobs(engine):
+    with engine.begin() as conn:
+        conn.execute(text("TRUNCATE evidence_jobs"))
+    yield api().EvidenceJobRepository(engine)
+    with engine.begin() as conn:
+        conn.execute(text("TRUNCATE evidence_jobs"))
+
+
+def seed(engine):
+    bundle = EvidenceBundle.model_validate(bundle_data())
+    snapshot, actor = uuid4(), uuid4()
+    EvidenceRepository(engine).save(snapshot, bundle)
+    identity = IdentityRepository(engine)
+    identity.register_identity(actor, "synthetic", str(actor))
+    identity.set_membership(actor, bundle.document.workspace_id, "owner")
+    return (
+        SessionRepository(engine).issue_session(actor).token,
+        bundle.document.workspace_id,
+        snapshot,
+    )
+
+
+def test_repository_exists():
+    assert api().EvidenceJobRepository
+
+
+def test_idempotency_claim_recovery_and_budget(jobs, engine):
+    token, workspace, snapshot = seed(engine)
+    request = uuid4()
+    job = jobs.submit(
+        token,
+        workspace,
+        request,
+        snapshot,
+        " exact ",
+        model="gpt-6.1-sol",
+        reservation_usd=Decimal("1.00"),
+    )
+    again = jobs.submit(
+        token,
+        workspace,
+        request,
+        snapshot,
+        " exact ",
+        model="gpt-6.1-sol",
+        reservation_usd=Decimal("1.00"),
+    )
+    assert again.job_id == job.job_id
+    with pytest.raises(api().ResearchConflict):
+        jobs.submit(
+            token,
+            workspace,
+            request,
+            snapshot,
+            "changed",
+            model="gpt-6.1-sol",
+            reservation_usd=Decimal("1.00"),
+        )
+    assert jobs.claim(job.job_id, prompt_sha256="a" * 64, instructions_sha256="b" * 64)
+    assert not jobs.claim(job.job_id, prompt_sha256="a" * 64, instructions_sha256="b" * 64)
+    assert jobs.recover(job.job_id).status == "indeterminate"
+    assert jobs.get(token, workspace, job.job_id).receipt is None
+    with engine.connect() as conn:
+        row = conn.execute(
+            text("SELECT reserved_usd,reservation_count FROM research_budget_months")
+        ).one()
+        assert row == (Decimal("1.00"), 1)
+
+
+@pytest.mark.parametrize(
+    "reservation", [1, 1.0, True, Decimal("NaN"), Decimal("0"), Decimal("10.01"), Decimal(".001")]
+)
+def test_exact_reservation(jobs, engine, reservation):
+    token, workspace, snapshot = seed(engine)
+    with pytest.raises(ValueError):
+        jobs.submit(
+            token,
+            workspace,
+            uuid4(),
+            snapshot,
+            "q",
+            model="gpt-6.1-sol",
+            reservation_usd=reservation,
+        )
+
+
+def test_source_hash_matches_agents(jobs, engine):
+    from issuedigest_agents.question_evidence_propositions import _source
+
+    token, workspace, snapshot = seed(engine)
+    job = jobs.submit(
+        token,
+        workspace,
+        uuid4(),
+        snapshot,
+        "q",
+        model="gpt-6.1-sol",
+        reservation_usd=Decimal(".25"),
+    )
+    _, source = jobs.prepare(job.job_id)
+    assert job.source_sha256 == _source(source)[0]
+
+
+def test_observations_over_cap_and_unknown_are_not_free(jobs, engine):
+    token, workspace, snapshot = seed(engine)
+    job = jobs.submit(
+        token,
+        workspace,
+        uuid4(),
+        snapshot,
+        "q",
+        model="gpt-6.1-sol",
+        reservation_usd=Decimal(".25"),
+    )
+    assert jobs.recover(job.job_id).known_usage is None
+    jobs.claim(job.job_id, prompt_sha256="a" * 64, instructions_sha256="b" * 64)
+    huge = dict(input_tokens=2**70, output_tokens=9000, requests=1)
+    failed = jobs.finish(
+        job.job_id, known_usage=huge, usage_complete=True, stream_closed=True, reason="output"
+    )
+    assert failed.known_usage == huge and failed.receipt is None and failed.status == "failed"
+    assert (
+        jobs.finish(
+            job.job_id, known_usage=dict(input_tokens=1, output_tokens=1, requests=1)
+        ).known_usage
+        == huge
+    )
+
+
+def test_budget_combines_old_pool_and_new(jobs, engine):
+    token, workspace, snapshot = seed(engine)
+    with engine.begin() as conn:
+        conn.execute(
+            text(
+                "INSERT INTO research_budget_months VALUES "
+                "(to_char(clock_timestamp(),'YYYY-MM'),9.75,39)"
+            )
+        )
+    job = jobs.submit(
+        token,
+        workspace,
+        uuid4(),
+        snapshot,
+        "q",
+        model="gpt-6.1-sol",
+        reservation_usd=Decimal(".25"),
+    )
+    assert job.status == "pending"
+    jobs.cancel(token, workspace, job.job_id)
+    with pytest.raises(api().ResearchBudgetExceeded):
+        jobs.submit(
+            token,
+            workspace,
+            uuid4(),
+            snapshot,
+            "q",
+            model="gpt-6.1-sol",
+            reservation_usd=Decimal(".01"),
+        )
+
+
+def test_viewer_can_read_current_job(jobs, engine):
+    token, workspace, snapshot = seed(engine)
+    job = jobs.submit(
+        token,
+        workspace,
+        uuid4(),
+        snapshot,
+        "q",
+        model="gpt-6.1-sol",
+        reservation_usd=Decimal(".25"),
+    )
+    viewer = uuid4()
+    identity = IdentityRepository(engine)
+    identity.register_identity(viewer, "synthetic", str(viewer))
+    identity.set_membership(viewer, workspace, "viewer")
+    viewer_token = SessionRepository(engine).issue_session(viewer).token
+    assert jobs.get(viewer_token, workspace, job.job_id).job_id == job.job_id
+    with pytest.raises(api().ResearchForbidden):
+        jobs.cancel(viewer_token, workspace, job.job_id)
+
+
+@pytest.fixture(scope="module", autouse=True)
+def restore_introduced_agent_modules():
+    import sys
+
+    package = importlib.import_module("issuedigest_agents")
+    existing, attributes = set(sys.modules), vars(package).copy()
+    yield
+    for name in set(sys.modules) - existing:
+        if name.startswith("issuedigest_agents."):
+            sys.modules.pop(name, None)
+    for name in set(vars(package)) - attributes.keys():
+        delattr(package, name)
+    for name, value in attributes.items():
+        setattr(package, name, value)
+
+
+def test_atomic_duplicate_claims(jobs, engine):
+    from concurrent.futures import ThreadPoolExecutor
+
+    token, workspace, snapshot = seed(engine)
+    job = jobs.submit(
+        token,
+        workspace,
+        uuid4(),
+        snapshot,
+        "q",
+        model="gpt-6.1-sol",
+        reservation_usd=Decimal(".25"),
+    )
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        results = list(
+            pool.map(
+                lambda _: jobs.claim(
+                    job.job_id, prompt_sha256="a" * 64, instructions_sha256="b" * 64
+                ),
+                range(2),
+            )
+        )
+    assert sorted(results) == [False, True]
+
+
+def test_migration_only_new_table_and_old_rows(jobs, engine, database_url):
+    from alembic import command
+    from sqlalchemy import inspect
+
+    from conftest import migration_config
+
+    token, workspace, snapshot = seed(engine)
+    with engine.connect() as conn:
+        old = conn.execute(text("SELECT snapshot_id,workspace_id FROM evidence_snapshots")).all()
+    config = migration_config(database_url)
+    before = set(inspect(engine).get_table_names())
+    command.downgrade(config, "0009_policy_intake")
+    try:
+        assert set(inspect(engine).get_table_names()) == before - {"evidence_jobs"}
+        with engine.connect() as conn:
+            assert (
+                conn.execute(text("SELECT snapshot_id,workspace_id FROM evidence_snapshots")).all()
+                == old
+            )
+    finally:
+        command.upgrade(config, "0010_evidence_jobs")
+    assert set(inspect(engine).get_table_names()) == before
+
+
+def test_source_metadata_change_rejects_same_request(jobs, engine):
+    from issuedigest_storage.repository import StorageCorruption
+    from issuedigest_storage.research import ResearchRepository
+
+    token, workspace, snapshot = seed(engine)
+    request = uuid4()
+    job = jobs.submit(
+        token,
+        workspace,
+        request,
+        snapshot,
+        "q",
+        model="gpt-6.1-sol",
+        reservation_usd=Decimal(".25"),
+    )
+    _, source = jobs.prepare(job.job_id)
+    ResearchRepository(engine).upsert_presentation(
+        source[2].model_copy(update={"title": "changed"})
+    )
+    with pytest.raises(api().ResearchConflict):
+        jobs.submit(
+            token,
+            workspace,
+            request,
+            snapshot,
+            "q",
+            model="gpt-6.1-sol",
+            reservation_usd=Decimal(".25"),
+        )
+    with pytest.raises(StorageCorruption):
+        jobs.prepare(job.job_id)
+
+
+@pytest.mark.parametrize(
+    "question,model",
+    [
+        ("", "gpt-6.1-sol"),
+        ("  ", "gpt-6.1-sol"),
+        ("q" * 1501, "gpt-6.1-sol"),
+        ("q", "gpt-5.4-mini"),
+        ("q", True),
+    ],
+)
+def test_exact_question_and_model(jobs, engine, question, model):
+    token, workspace, snapshot = seed(engine)
+    with pytest.raises(ValueError):
+        jobs.submit(
+            token,
+            workspace,
+            uuid4(),
+            snapshot,
+            question,
+            model=model,
+            reservation_usd=Decimal(".25"),
+        )
+
+
+def test_invalid_session_other_workspace_and_missing_source(jobs, engine):
+    from issuedigest_storage.sessions import AuthenticationDenied
+
+    token, workspace, snapshot = seed(engine)
+    with pytest.raises(AuthenticationDenied):
+        jobs.submit(
+            "invalid",
+            workspace,
+            uuid4(),
+            snapshot,
+            "q",
+            model="gpt-6.1-sol",
+            reservation_usd=Decimal(".25"),
+        )
+    assert (
+        jobs.submit(
+            token,
+            uuid4(),
+            uuid4(),
+            snapshot,
+            "q",
+            model="gpt-6.1-sol",
+            reservation_usd=Decimal(".25"),
+        )
+        is None
+    )
+    with pytest.raises(api().ResearchConflict):
+        jobs.submit(
+            token,
+            workspace,
+            uuid4(),
+            uuid4(),
+            "q",
+            model="gpt-6.1-sol",
+            reservation_usd=Decimal(".25"),
+        )
+
+
+def test_concurrent_shared_budget_last_reservation(jobs, engine):
+    from concurrent.futures import ThreadPoolExecutor
+
+    one = seed(engine)
+    two = seed(engine)
+    with engine.begin() as conn:
+        conn.execute(
+            text(
+                "INSERT INTO research_budget_months VALUES "
+                "(to_char(clock_timestamp(),'YYYY-MM'),9.75,39)"
+            )
+        )
+
+    def submit_one(seed_data):
+        token, workspace, snapshot = seed_data
+        try:
+            return jobs.submit(
+                token,
+                workspace,
+                uuid4(),
+                snapshot,
+                "q",
+                model="gpt-6.1-sol",
+                reservation_usd=Decimal(".25"),
+            ).status
+        except api().ResearchBudgetExceeded:
+            return "budget"
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        assert sorted(pool.map(submit_one, [one, two])) == ["budget", "pending"]
+
+
+def test_finish_storage_rejects_mismatched_source_identity(jobs, engine):
+    from issuedigest_storage.repository import StorageCorruption
+
+    token, workspace, snapshot = seed(engine)
+    job = jobs.submit(
+        token,
+        workspace,
+        uuid4(),
+        snapshot,
+        "q",
+        model="gpt-6.1-sol",
+        reservation_usd=Decimal(".25"),
+    )
+    jobs.claim(job.job_id, prompt_sha256="a" * 64, instructions_sha256="b" * 64)
+    usage = dict(input_tokens=10, output_tokens=5, requests=1)
+    fake = dict(
+        profile="guarded-question-evidence-v2",
+        status="unreviewed",
+        prompt_sha256="a" * 64,
+        instructions_sha256="b" * 64,
+        usage=usage,
+        evidence={"source_sha256": "f" * 64},
+    )
+    with pytest.raises(StorageCorruption):
+        jobs.finish(
+            job.job_id, receipt=fake, known_usage=usage, usage_complete=True, stream_closed=True
+        )
+    assert jobs.prepare(job.job_id)[0].status == "running"
+
+
+def test_actual_old_run_and_new_job_share_pool(jobs, engine):
+    from issuedigest_domain.research_journey import ResearchRequest
+    from issuedigest_storage.research import ResearchRepository
+
+    token, workspace, snapshot = seed(engine)
+    old = ResearchRepository(engine).submit(
+        token, workspace, ResearchRequest(request_id=uuid4(), question="old")
+    )
+    assert old is not None
+    job = jobs.submit(
+        token,
+        workspace,
+        uuid4(),
+        snapshot,
+        "new",
+        model="gpt-6.1-sol",
+        reservation_usd=Decimal("9.75"),
+    )
+    assert job.status == "pending"
+    with engine.connect() as conn:
+        assert conn.execute(
+            text("SELECT reserved_usd,reservation_count FROM research_budget_months")
+        ).one() == (Decimal("10"), 2)
+
+
+@pytest.mark.parametrize("change", ["revoke", "title", "delete"])
+def test_completed_cancel_revalidates_original_authority_and_source(jobs, engine, change):
+    from issuedigest_storage.auth_schema import auth_workspace_memberships
+    from issuedigest_storage.repository import StorageCorruption
+    from issuedigest_storage.research import ResearchConflict, ResearchForbidden, ResearchRepository
+    from sqlalchemy import update
+
+    token, workspace, snapshot = seed(engine)
+    job = jobs.submit(
+        token,
+        workspace,
+        uuid4(),
+        snapshot,
+        "q",
+        model="gpt-6.1-sol",
+        reservation_usd=Decimal(".25"),
+    )
+    receipt = completion_receipt(jobs, job)
+    jobs.claim(
+        job.job_id,
+        prompt_sha256=receipt["prompt_sha256"],
+        instructions_sha256=receipt["instructions_sha256"],
+    )
+    jobs.finish(
+        job.job_id,
+        receipt=receipt,
+        known_usage=receipt["usage"],
+        usage_complete=True,
+        stream_closed=True,
+    )
+    editor = uuid4()
+    identity = IdentityRepository(engine)
+    identity.register_identity(editor, "synthetic", str(editor))
+    identity.set_membership(editor, workspace, "editor")
+    editor_token = SessionRepository(engine).issue_session(editor).token
+    if change == "revoke":
+        with engine.begin() as conn:
+            conn.execute(
+                update(auth_workspace_memberships)
+                .where(auth_workspace_memberships.c.user_id != editor)
+                .values(active=False)
+            )
+    elif change == "title":
+        _, source = jobs.prepare(job.job_id)
+        ResearchRepository(engine).upsert_presentation(
+            source[2].model_copy(update={"title": "changed"})
+        )
+    else:
+        with engine.begin() as conn:
+            for table in ("snapshot_citations", "outbox"):
+                conn.execute(
+                    text("DELETE FROM " + table + " WHERE snapshot_id=:sid"), {"sid": snapshot}
+                )
+            conn.execute(
+                text("DELETE FROM evidence_snapshots WHERE snapshot_id=:sid"), {"sid": snapshot}
+            )
+    with pytest.raises((ResearchForbidden, ResearchConflict, StorageCorruption)):
+        jobs.cancel(editor_token, workspace, job.job_id)
+
+
+def completion_receipt(jobs, job):
+    from hashlib import sha256
+
+    from issuedigest_agents.guarded_question_evidence_v2 import (
+        GUARDED_QUESTION_V2_INSTRUCTIONS,
+        GuardedQuestionEvidenceGenerationV2,
+        _preflight,
+    )
+    from issuedigest_agents.question_evidence_propositions import (
+        build_question_evidence_propositions,
+    )
+
+    _, source = jobs.prepare(job.job_id)
+    prompt, _ = _preflight(source, job.question)
+    evidence = build_question_evidence_propositions(
+        source,
+        job.question,
+        dict(
+            propositions=[],
+            coverage=dict(linked_proposition_ids=[], unresolved_reason="insufficient"),
+        ),
+    )
+    return GuardedQuestionEvidenceGenerationV2(
+        profile="guarded-question-evidence-v2",
+        transport_profile="guarded-responses-tool-v1",
+        input_format="question-evidence-source-v2",
+        status="unreviewed",
+        evidence=evidence,
+        prompt_sha256=sha256(prompt.encode()).hexdigest(),
+        instructions_sha256=sha256(GUARDED_QUESTION_V2_INSTRUCTIONS.encode()).hexdigest(),
+        usage=dict(input_tokens=10, output_tokens=5, requests=1),
+    ).model_dump(mode="json")
+
+
+@pytest.mark.parametrize("public_operation", ["get", "cancel"])
+@pytest.mark.parametrize("worker_operation", ["claim", "finish"])
+def test_authenticated_worker_lock_order_race(jobs, engine, public_operation, worker_operation):
+    from concurrent.futures import ThreadPoolExecutor
+    from threading import Event, get_ident
+    from time import monotonic
+
+    from sqlalchemy import event
+
+    token, workspace, snapshot = seed(engine)
+    job = jobs.submit(
+        token,
+        workspace,
+        uuid4(),
+        snapshot,
+        "q",
+        model="gpt-6.1-sol",
+        reservation_usd=Decimal(".25"),
+    )
+    receipt = completion_receipt(jobs, job)
+    if worker_operation == "finish":
+        assert jobs.claim(
+            job.job_id,
+            prompt_sha256=receipt["prompt_sha256"],
+            instructions_sha256=receipt["instructions_sha256"],
+        )
+    public_locked, release_public, worker_started = Event(), Event(), Event()
+    ids = {}
+
+    def before(conn, cursor, statement, parameters, context, executemany):
+        if get_ident() == ids.get("worker_thread"):
+            ids.setdefault("worker_pid", conn.connection.driver_connection.info.backend_pid)
+            worker_started.set()
+
+    def after(conn, cursor, statement, parameters, context, executemany):
+        if (
+            get_ident() == ids.get("public_thread")
+            and "auth_workspace_memberships" in statement
+            and "FOR UPDATE" in statement
+            and not public_locked.is_set()
+        ):
+            ids["public_pid"] = conn.connection.driver_connection.info.backend_pid
+            public_locked.set()
+            assert release_public.wait(10)
+
+    def public():
+        ids["public_thread"] = get_ident()
+        return getattr(jobs, public_operation)(token, workspace, job.job_id)
+
+    def worker():
+        ids["worker_thread"] = get_ident()
+        if worker_operation == "claim":
+            return jobs.claim(
+                job.job_id,
+                prompt_sha256=receipt["prompt_sha256"],
+                instructions_sha256=receipt["instructions_sha256"],
+            )
+        return jobs.finish(
+            job.job_id,
+            receipt=receipt,
+            known_usage=receipt["usage"],
+            usage_complete=True,
+            stream_closed=True,
+        )
+
+    event.listen(engine, "before_cursor_execute", before)
+    event.listen(engine, "after_cursor_execute", after)
+    try:
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            first = pool.submit(public)
+            try:
+                assert public_locked.wait(5)
+                second = pool.submit(worker)
+                assert worker_started.wait(5)
+                with engine.connect() as observer:
+                    deadline = monotonic() + 5
+                    while ids["public_pid"] not in observer.scalar(
+                        text("SELECT pg_blocking_pids(:pid)"), {"pid": ids["worker_pid"]}
+                    ):
+                        assert monotonic() < deadline, (
+                            "worker did not block behind public transaction"
+                        )
+                        Event().wait(0.01)
+                release_public.set()
+                assert first.result(timeout=10) is not None
+                second.result(timeout=10)
+            finally:
+                release_public.set()
+    finally:
+        event.remove(engine, "before_cursor_execute", before)
+        event.remove(engine, "after_cursor_execute", after)
+    state = jobs.prepare(job.job_id)[0].status
+    assert state == (
+        "cancelled"
+        if public_operation == "cancel"
+        else "running"
+        if worker_operation == "claim"
+        else "completed"
+    )