From 4e8495e8df5b2d356850c131d908090637b26921 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 7 Sep 2026 20:46:51 +0900 Subject: [PATCH 01/23] fix(ask): fence Ask settlement on the claim generation Orphan recovery can flip a long-running job back to queued while the original worker still holds the answer. Settlement now compare-and-sets the claim's updated_at so the previous owner cannot overwrite a reclaimed row. --- backend/app/global_ask_queue.py | 17 +++++- docs/product-technical-gap-baseline.md | 7 +++ tests/test_global_ask_queue.py | 79 ++++++++++++++++++++++++-- 3 files changed, 96 insertions(+), 7 deletions(-) diff --git a/backend/app/global_ask_queue.py b/backend/app/global_ask_queue.py index f8720f715..19acbbe1e 100644 --- a/backend/app/global_ask_queue.py +++ b/backend/app/global_ask_queue.py @@ -520,8 +520,10 @@ async def process_global_ask_job( Claiming flips ``queued`` → ``running`` atomically so a duplicate stream wake-up (recovery republish racing the original entry) is a - no-op. Every failure path settles the row as ``failed`` with a - bounded detail string rather than leaving it stuck ``running``. + no-op. Settlement is compare-and-set on that claim's ``updated_at`` + so an orphan reclaim cannot be overwritten by the previous owner. + Every failure path settles the row as ``failed`` with a bounded + detail string rather than leaving it stuck ``running``. """ async with pool.acquire() as conn: row = await conn.fetchrow( @@ -529,7 +531,7 @@ async def process_global_ask_job( update global_ask_job set job_status_code = $2, updated_at = now() where global_ask_job_id = $1 and job_status_code = $3 returning requesting_account_id, question_text, verify_external_requested, - knowledge_cutoff + knowledge_cutoff, updated_at """, job_id, RUNNING, @@ -537,6 +539,7 @@ async def process_global_ask_job( ) if row is None: return + claimed_at = row["updated_at"] answer_timeout: asyncio.Timeout | None = None try: async with pool.acquire() as conn: @@ -607,10 +610,14 @@ async def process_global_ask_job( update global_ask_job set job_status_code = $2, failure_detail = $3, updated_at = now() where global_ask_job_id = $1 + and job_status_code = $4 + and updated_at = $5 """, job_id, FAILED, detail[:1000], + RUNNING, + claimed_at, ) return async with pool.acquire() as conn: @@ -619,10 +626,14 @@ async def process_global_ask_job( update global_ask_job set job_status_code = $2, answer_payload = $3::jsonb, updated_at = now() where global_ask_job_id = $1 + and job_status_code = $4 + and updated_at = $5 """, job_id, SUCCEEDED, _to_json(payload), + RUNNING, + claimed_at, ) diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 42dc5d155..1b6e0fd22 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -1,5 +1,12 @@ # Product & Technical Gap Baseline +> Ask ownership-fence overlay: 2026-09-07 KST. Issue #975 stacks onto +> #974 so Global Ask settlement is compare-and-set on the claim +> generation (`updated_at`). Orphan recovery can still reclaim by age; +> the previous owner cannot overwrite the reclaimed row. The 600 s +> worker deadline remains until liveness heartbeat lands. Draft; not +> protected-main or independently approved evidence. +> > Exact-head loop overlay: 2026-08-29 13:20 KST. Protected `main` is > `fc13acaa20adca11968238e398d4aafcf62b6cee` (v2.23.0 leftover-map > explained leftover share, #775). Open ready PRs still lack independent diff --git a/tests/test_global_ask_queue.py b/tests/test_global_ask_queue.py index 453a933ef..10bf36c9a 100644 --- a/tests/test_global_ask_queue.py +++ b/tests/test_global_ask_queue.py @@ -40,12 +40,16 @@ async def acquire(self): yield self.connection +_CLAIMED_AT = datetime(2026, 1, 1, 9, 0, tzinfo=UTC) + + def _queued_row() -> dict[str, object]: return { "requesting_account_id": "00000000-0000-0000-0000-000000000001", "question_text": "What happened last week?", "verify_external_requested": False, "knowledge_cutoff": None, + "updated_at": _CLAIMED_AT, } @@ -398,7 +402,7 @@ async def _fake_compute_global_ask_answer(*_args, **_kwargs): settle_query, settle_args = connection.executed[-1] assert "failure_detail" in settle_query - failure_detail = settle_args[-1] + failure_detail = settle_args[2] assert secret_bearing_message not in failure_detail assert failure_detail == ( "Ask Agent is unavailable: contextual-orchestrator returned no complete evidence object" @@ -434,7 +438,8 @@ async def _fake_compute_global_ask_answer(*_args, **_kwargs): ) _settle_query, settle_args = connection.executed[-1] - assert settle_args[-1] == "account lacks the post_read permission" + assert settle_args[2] == "account lacks the post_read permission" + assert settle_args[3:] == (global_ask_queue.RUNNING, _CLAIMED_AT) @pytest.mark.parametrize("timeout_source", ["provider", "worker", "shutdown"]) @@ -480,12 +485,78 @@ async def _fake_compute_global_ask_answer(*_args, **_kwargs): _settle_query, settle_args = connection.executed[-1] if timeout_source == "worker": - assert settle_args[-1] == "job exceeded the 0s deadline" + assert settle_args[2] == "job exceeded the 0s deadline" else: - assert settle_args[-1] == ( + assert settle_args[2] == ( "Ask Agent is unavailable: contextual-orchestrator returned " "no complete evidence object" ) + assert settle_args[3:] == (global_ask_queue.RUNNING, _CLAIMED_AT) + + +def test_orphan_reclaim_prevents_stale_owner_from_settling(monkeypatch) -> None: + """After recovery reclaims a running job, the previous owner cannot settle it.""" + reclaimed_at = datetime(2026, 1, 2, 9, 0, tzinfo=UTC) + + class _RaceConnection(_Connection): + def __init__(self) -> None: + super().__init__(_queued_row()) + self.status = global_ask_queue.QUEUED + self.generation = _CLAIMED_AT + self.applied: list[str] = [] + + async def fetchrow(self, query: str, *_args: object): + if "job_status_code = $3" in query and self.status == global_ask_queue.QUEUED: + self.status = global_ask_queue.RUNNING + return self.row + return None + + async def execute(self, query: str, *args: object) -> str: + self.executed.append((query, args)) + if "answer_payload" in query: + claimed = args[4] if len(args) > 4 else None + if ( + "updated_at = $5" in query + and args[3] == global_ask_queue.RUNNING + and claimed == self.generation + and self.status == global_ask_queue.RUNNING + ): + self.status = str(args[1]) + self.applied.append("accepted") + return "UPDATE 1" + self.applied.append("rejected") + return "UPDATE 0" + return "OK" + + connection = _RaceConnection() + pool = _Pool(connection) + + async def _fake_load_job_visibility(_conn, _job_id, _account_id): + return {"corp-1"}, set(), False, True + + async def _fake_compute_global_ask_answer(*_args, **_kwargs): + connection.status = global_ask_queue.RUNNING + connection.generation = reclaimed_at + return {"answer_text": "stale-owner-payload"} + + monkeypatch.setattr(global_ask_queue, "load_job_visibility", _fake_load_job_visibility) + monkeypatch.setattr( + global_ask_queue, "compute_global_ask_answer", _fake_compute_global_ask_answer + ) + + asyncio.run( + global_ask_queue.process_global_ask_job( + pool, + job_id="job-1", + chat_factory=_AvailableClient, + ) + ) + + settle_query, settle_args = connection.executed[-1] + assert "updated_at = $5" in settle_query + assert settle_args[3:] == (global_ask_queue.RUNNING, _CLAIMED_AT) + assert connection.applied == ["rejected"] + assert connection.status == global_ask_queue.RUNNING def test_job_visibility_never_expands_past_queued_scope() -> None: From ba5a6fb254fe351416e0d85154a007d4df7f1897 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 7 Sep 2026 21:45:40 +0900 Subject: [PATCH 02/23] fix(ask): treat a lost claim generation as an unapplied settle Compare-and-set settlement now inspects PostgreSQL UPDATE 0 so a reclaimed Ask job is left with the new owner instead of being misread as a completed write. --- backend/app/global_ask_queue.py | 13 +++++++++++-- docs/product-technical-gap-baseline.md | 9 +++++---- tests/test_global_ask_queue.py | 4 +++- 3 files changed, 19 insertions(+), 7 deletions(-) diff --git a/backend/app/global_ask_queue.py b/backend/app/global_ask_queue.py index 19acbbe1e..7835d31c6 100644 --- a/backend/app/global_ask_queue.py +++ b/backend/app/global_ask_queue.py @@ -102,6 +102,11 @@ class _SafeJobError(Exception): """Failure whose bounded message is safe to persist for the requester.""" +def _claim_generation_retained(command_status: object) -> bool: + """Return True when PostgreSQL reports the compare-and-set updated one row.""" + return str(command_status) == "UPDATE 1" + + async def enqueue_global_ask_job( conn: asyncpg.Connection, client: redis.Redis, @@ -605,7 +610,7 @@ async def process_global_ask_job( "no complete evidence object" ) async with pool.acquire() as conn: - await conn.execute( + command_status = await conn.execute( """ update global_ask_job set job_status_code = $2, failure_detail = $3, updated_at = now() @@ -619,9 +624,11 @@ async def process_global_ask_job( RUNNING, claimed_at, ) + if not _claim_generation_retained(command_status): + return return async with pool.acquire() as conn: - await conn.execute( + command_status = await conn.execute( """ update global_ask_job set job_status_code = $2, answer_payload = $3::jsonb, updated_at = now() @@ -635,6 +642,8 @@ async def process_global_ask_job( RUNNING, claimed_at, ) + if not _claim_generation_retained(command_status): + return def _to_json(payload: dict[str, Any]) -> str: diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 1b6e0fd22..64b321be8 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -2,10 +2,11 @@ > Ask ownership-fence overlay: 2026-09-07 KST. Issue #975 stacks onto > #974 so Global Ask settlement is compare-and-set on the claim -> generation (`updated_at`). Orphan recovery can still reclaim by age; -> the previous owner cannot overwrite the reclaimed row. The 600 s -> worker deadline remains until liveness heartbeat lands. Draft; not -> protected-main or independently approved evidence. +> generation (`updated_at`). A PostgreSQL `UPDATE 0` means the previous +> owner lost the generation and is not a buyer-visible failure. Orphan +> recovery can still reclaim by age. The 600 s worker deadline remains +> until liveness heartbeat lands. Draft; not protected-main or +> independently approved evidence. > > Exact-head loop overlay: 2026-08-29 13:20 KST. Protected `main` is > `fc13acaa20adca11968238e398d4aafcf62b6cee` (v2.23.0 leftover-map diff --git a/tests/test_global_ask_queue.py b/tests/test_global_ask_queue.py index 10bf36c9a..850fa73ce 100644 --- a/tests/test_global_ask_queue.py +++ b/tests/test_global_ask_queue.py @@ -28,7 +28,7 @@ async def fetchrow(self, *_args: object): async def execute(self, query: str, *args: object) -> str: self.executed.append((query, args)) - return "OK" + return "UPDATE 1" class _Pool: @@ -557,6 +557,8 @@ async def _fake_compute_global_ask_answer(*_args, **_kwargs): assert settle_args[3:] == (global_ask_queue.RUNNING, _CLAIMED_AT) assert connection.applied == ["rejected"] assert connection.status == global_ask_queue.RUNNING + assert global_ask_queue._claim_generation_retained("UPDATE 0") is False + assert global_ask_queue._claim_generation_retained("UPDATE 1") is True def test_job_visibility_never_expands_past_queued_scope() -> None: From 3177599f48f6f1b919e47b1a9dfe592401104aef Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 7 Sep 2026 22:46:37 +0900 Subject: [PATCH 03/23] fix(ask): renew Ask claim generation while the worker is live Age-based orphan recovery could reclaim a job whose owner was still computing. Renew updated_at on the recovery interval, abort when the generation is lost, and keep the 600 s deadline until that heartbeat is independently approved. --- backend/app/global_ask_queue.py | 101 +++++++++++++++++++++---- docs/product-technical-gap-baseline.md | 6 +- tests/test_global_ask_queue.py | 83 +++++++++++++++++++- 3 files changed, 173 insertions(+), 17 deletions(-) diff --git a/backend/app/global_ask_queue.py b/backend/app/global_ask_queue.py index 7835d31c6..9c5b1744c 100644 --- a/backend/app/global_ask_queue.py +++ b/backend/app/global_ask_queue.py @@ -94,6 +94,9 @@ # LLM round-trips, so serial consumption would head-of-line block every # later question behind the slowest one. _WORKER_CONCURRENCY = 4 +# Live workers renew the claim generation on this interval so age-based +# orphan recovery cannot reclaim a job that is still owned. +_CLAIM_HEARTBEAT_SECONDS = _RECOVERY_INTERVAL_SECONDS _logger = logging.getLogger(__name__) @@ -107,6 +110,69 @@ def _claim_generation_retained(command_status: object) -> bool: return str(command_status) == "UPDATE 1" +class _LostAskClaim(Exception): + """Raised when orphan recovery reclaims the running claim generation.""" + + +async def _renew_ask_claim( + pool: asyncpg.Pool, job_id: str, claimed_at: object +) -> object | None: + """Advance ``updated_at`` only for the current running claim generation.""" + async with pool.acquire() as conn: + row = await conn.fetchrow( + """ + update global_ask_job set updated_at = now() + where global_ask_job_id = $1 + and job_status_code = $2 + and updated_at = $3 + returning updated_at + """, + job_id, + RUNNING, + claimed_at, + ) + if row is None: + return None + return row["updated_at"] + + +async def _run_with_ask_claim_heartbeat( + pool: asyncpg.Pool, + job_id: str, + lease: list[object], + operation: Any, +) -> Any: + """Renew the claim generation while ``operation`` runs; abort on reclaim.""" + lost = asyncio.Event() + stop = asyncio.Event() + + async def beat() -> None: + while not stop.is_set() and not lost.is_set(): + try: + await asyncio.wait_for(stop.wait(), timeout=_CLAIM_HEARTBEAT_SECONDS) + return + except TimeoutError: + renewed = await _renew_ask_claim(pool, job_id, lease[0]) + if renewed is None: + lost.set() + return + lease[0] = renewed + + beater = asyncio.create_task(beat()) + worker = asyncio.create_task(operation) + try: + await asyncio.wait({worker, beater}, return_when=asyncio.FIRST_COMPLETED) + if lost.is_set(): + worker.cancel() + await asyncio.gather(worker, return_exceptions=True) + raise _LostAskClaim() + return await worker + finally: + stop.set() + beater.cancel() + await asyncio.gather(beater, return_exceptions=True) + + async def enqueue_global_ask_job( conn: asyncpg.Connection, client: redis.Redis, @@ -544,7 +610,7 @@ async def process_global_ask_job( ) if row is None: return - claimed_at = row["updated_at"] + lease = [row["updated_at"]] answer_timeout: asyncio.Timeout | None = None try: async with pool.acquire() as conn: @@ -564,19 +630,26 @@ async def process_global_ask_job( "Ask Agent is unavailable: set ORCHESTRATOR_BASE_URL / ORCHESTRATOR_API_KEY" ) async with asyncio.timeout(JOB_DEADLINE_SECONDS) as answer_timeout: - payload = await compute_global_ask_answer( + payload = await _run_with_ask_claim_heartbeat( pool, - question_text=str(row["question_text"]), - corporate_entity_ids=entity_ids, - process_unit_ids=process_unit_ids, - process_scope_limited=process_scope_limited, - chat_client=chat_client, - embedding_client=embedding_factory(), - semantic_query_client=semantic_query_factory(), - verify_external=bool(row["verify_external_requested"]), - claim_verification_client=claim_verification_factory(), - knowledge_cutoff=row["knowledge_cutoff"], + job_id, + lease, + compute_global_ask_answer( + pool, + question_text=str(row["question_text"]), + corporate_entity_ids=entity_ids, + process_unit_ids=process_unit_ids, + process_scope_limited=process_scope_limited, + chat_client=chat_client, + embedding_client=embedding_factory(), + semantic_query_client=semantic_query_factory(), + verify_external=bool(row["verify_external_requested"]), + claim_verification_client=claim_verification_factory(), + knowledge_cutoff=row["knowledge_cutoff"], + ), ) + except _LostAskClaim: + return except asyncio.CancelledError: # Shutdown: leave the row `running`; the recovery sweep re-queues # it after the orphan window on the next process start. @@ -622,7 +695,7 @@ async def process_global_ask_job( FAILED, detail[:1000], RUNNING, - claimed_at, + lease[0], ) if not _claim_generation_retained(command_status): return @@ -640,7 +713,7 @@ async def process_global_ask_job( SUCCEEDED, _to_json(payload), RUNNING, - claimed_at, + lease[0], ) if not _claim_generation_retained(command_status): return diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 64b321be8..25f4f020d 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -4,8 +4,10 @@ > #974 so Global Ask settlement is compare-and-set on the claim > generation (`updated_at`). A PostgreSQL `UPDATE 0` means the previous > owner lost the generation and is not a buyer-visible failure. Orphan -> recovery can still reclaim by age. The 600 s worker deadline remains -> until liveness heartbeat lands. Draft; not protected-main or +> recovery can still reclaim by age. Live workers renew ``updated_at`` +> on the recovery interval so a current owner is not reclaimed by +> elapsed time. The 600 s worker deadline remains until that heartbeat +> evidence is independently approved. Draft; not protected-main or > independently approved evidence. > > Exact-head loop overlay: 2026-08-29 13:20 KST. Protected `main` is diff --git a/tests/test_global_ask_queue.py b/tests/test_global_ask_queue.py index 850fa73ce..4a82b3ca5 100644 --- a/tests/test_global_ask_queue.py +++ b/tests/test_global_ask_queue.py @@ -4,7 +4,7 @@ import asyncio from contextlib import asynccontextmanager -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta import pytest @@ -561,6 +561,87 @@ async def _fake_compute_global_ask_answer(*_args, **_kwargs): assert global_ask_queue._claim_generation_retained("UPDATE 1") is True +def test_claim_heartbeat_advances_generation_used_at_settle(monkeypatch) -> None: + """A live worker renews updated_at so settlement matches the current generation.""" + renewed_at = _CLAIMED_AT + timedelta(seconds=30) + + class _HeartbeatConnection(_Connection): + async def fetchrow(self, query: str, *_args: object): + if "knowledge_cutoff, updated_at" in query: + return self.row + if "returning updated_at" in query: + assert self.row is not None + self.row = {**self.row, "updated_at": renewed_at} + return {"updated_at": renewed_at} + return self.row + + connection = _HeartbeatConnection(_queued_row()) + pool = _Pool(connection) + + async def _fake_load_job_visibility(_conn, _job_id, _account_id): + return {"corp-1"}, set(), False, True + + async def _fake_compute_global_ask_answer(*_args, **_kwargs): + await asyncio.sleep(0.05) + return {"answer_text": "live-owner"} + + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.01) + monkeypatch.setattr(global_ask_queue, "load_job_visibility", _fake_load_job_visibility) + monkeypatch.setattr( + global_ask_queue, "compute_global_ask_answer", _fake_compute_global_ask_answer + ) + + asyncio.run( + global_ask_queue.process_global_ask_job( + pool, + job_id="job-1", + chat_factory=_AvailableClient, + ) + ) + + settle_query, settle_args = connection.executed[-1] + assert "answer_payload" in settle_query + assert settle_args[4] == renewed_at + + +def test_lost_heartbeat_does_not_settle_a_reclaimed_job(monkeypatch) -> None: + """A failed claim renew leaves the reclaimed owner in place.""" + + class _LostHeartbeatConnection(_Connection): + async def fetchrow(self, query: str, *_args: object): + if "knowledge_cutoff, updated_at" in query: + return self.row + if "returning updated_at" in query: + return None + return self.row + + connection = _LostHeartbeatConnection(_queued_row()) + pool = _Pool(connection) + + async def _fake_load_job_visibility(_conn, _job_id, _account_id): + return {"corp-1"}, set(), False, True + + async def _fake_compute_global_ask_answer(*_args, **_kwargs): + await asyncio.Event().wait() + return {"answer_text": "should-not-settle"} + + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.01) + monkeypatch.setattr(global_ask_queue, "load_job_visibility", _fake_load_job_visibility) + monkeypatch.setattr( + global_ask_queue, "compute_global_ask_answer", _fake_compute_global_ask_answer + ) + + asyncio.run( + global_ask_queue.process_global_ask_job( + pool, + job_id="job-1", + chat_factory=_AvailableClient, + ) + ) + + assert all("answer_payload" not in query for query, _args in connection.executed) + + def test_job_visibility_never_expands_past_queued_scope() -> None: """The worker uses stored scope rows, not every account affiliation.""" From ee4430fa20f5cfe9c0d623061d92bc662c3353b8 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 7 Sep 2026 22:59:37 +0900 Subject: [PATCH 04/23] test(ask): reproduce detached worker on owner cancellation --- tests/test_global_ask_claim_cancellation.py | 44 +++++++++++++++++++++ 1 file changed, 44 insertions(+) create mode 100644 tests/test_global_ask_claim_cancellation.py diff --git a/tests/test_global_ask_claim_cancellation.py b/tests/test_global_ask_claim_cancellation.py new file mode 100644 index 000000000..a591e70b7 --- /dev/null +++ b/tests/test_global_ask_claim_cancellation.py @@ -0,0 +1,44 @@ +"""Cancellation contracts for the Global Ask claim-heartbeat boundary.""" + +from __future__ import annotations + +import asyncio + +import pytest + +from backend.app import global_ask_queue + + +def test_external_cancellation_reaches_active_ask_operation(monkeypatch) -> None: + """Cancelling the owner task must not leave its Ask operation detached.""" + operation_started = asyncio.Event() + operation_cancelled = asyncio.Event() + + async def operation() -> None: + operation_started.set() + try: + await asyncio.Event().wait() + finally: + operation_cancelled.set() + + async def exercise() -> None: + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 3600.0) + owner = asyncio.create_task( + global_ask_queue._run_with_ask_claim_heartbeat( + object(), + "job-1", + [object()], + operation(), + ) + ) + await operation_started.wait() + owner.cancel() + with pytest.raises(asyncio.CancelledError): + await owner + await asyncio.sleep(0) + assert operation_cancelled.is_set(), ( + "owner cancellation left compute_global_ask_answer detached from the " + "claim heartbeat lifecycle" + ) + + asyncio.run(exercise()) From 3f0b63d9782cbcc2c5e8e803cd882d2c36a81649 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 7 Sep 2026 23:06:20 +0900 Subject: [PATCH 05/23] ci(ask): run test-first cancellation repair for #979 --- .../automation-979-cancellation-repair.yml | 154 ++++++++++++++++++ 1 file changed, 154 insertions(+) create mode 100644 .github/workflows/automation-979-cancellation-repair.yml diff --git a/.github/workflows/automation-979-cancellation-repair.yml b/.github/workflows/automation-979-cancellation-repair.yml new file mode 100644 index 000000000..2fd7fb923 --- /dev/null +++ b/.github/workflows/automation-979-cancellation-repair.yml @@ -0,0 +1,154 @@ +name: Automation 979 cancellation repair + +on: + push: + branches: + - feat/ask-ownership-fenced-liveness-20260907 + +permissions: + contents: write + pull-requests: read + +concurrency: + group: automation-979-cancellation-repair + cancel-in-progress: false + +jobs: + repair: + if: github.actor != 'github-actions[bot]' + runs-on: ubuntu-24.04 + services: + postgres: + image: postgres:16-alpine@sha256:57c72fd2a128e416c7fcc499958864df5301e940bca0a56f58fddf30ffc07777 + env: + POSTGRES_PASSWORD: postgres + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U postgres" + --health-interval 10s + --health-timeout 5s + --health-retries 5 + env: + LINEAGEWEAVE_TEST_POSTGRES_ADMIN_DSN: postgresql://postgres:postgres@localhost:5432/postgres + EXPECTED_PRODUCT_PARENT: ee4430fa20f5cfe9c0d623061d92bc662c3353b8 + PRODUCT_BRANCH: feat/ask-ownership-fenced-liveness-20260907 + steps: + - name: Checkout exact automation head + uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # actions/checkout@v7 + with: + fetch-depth: 0 + + - name: Verify authority before repair + env: + GH_TOKEN: ${{ github.token }} + run: | + set -euo pipefail + test "$(git rev-parse HEAD^)" = "$EXPECTED_PRODUCT_PARENT" + live_head="$(gh api repos/${GITHUB_REPOSITORY}/pulls/979 --jq .head.sha)" + test "$live_head" = "$GITHUB_SHA" + test "$(git status --porcelain)" = "" + + - name: Set up Python + uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # actions/setup-python@v6 + with: + python-version: "3.12" + + - name: Set up locked dependency manager + uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0 + with: + version: "0.11.28" + enable-cache: false + + - name: Select pinned Rust toolchain + run: | + rustup toolchain install 1.97.1 --profile minimal + rustup default 1.97.1 + + - name: Install committed dependency graph + run: uv sync --frozen --extra dev --extra backend + + - name: Reproduce external-cancellation RED + run: | + set -euo pipefail + set +e + uv run --frozen python -m pytest -q tests/test_global_ask_claim_cancellation.py > /tmp/red.log 2>&1 + red_status=$? + set -e + cat /tmp/red.log + test "$red_status" -ne 0 + grep -F "owner cancellation left compute_global_ask_answer detached" /tmp/red.log + + - name: Apply minimum causal cancellation repair + run: | + set -euo pipefail + python - <<'PY' + from pathlib import Path + + path = Path("backend/app/global_ask_queue.py") + text = path.read_text() + old = ''' beater = asyncio.create_task(beat()) + worker = asyncio.create_task(operation) + try: + await asyncio.wait({worker, beater}, return_when=asyncio.FIRST_COMPLETED) + if lost.is_set(): + worker.cancel() + await asyncio.gather(worker, return_exceptions=True) + raise _LostAskClaim() + return await worker + finally: + stop.set() + beater.cancel() + await asyncio.gather(beater, return_exceptions=True) + ''' + new = ''' beater = asyncio.create_task(beat()) + worker = asyncio.create_task(operation) + try: + await asyncio.wait({worker, beater}, return_when=asyncio.FIRST_COMPLETED) + if lost.is_set(): + worker.cancel() + await asyncio.gather(worker, return_exceptions=True) + raise _LostAskClaim() + return await worker + finally: + stop.set() + for task in (beater, worker): + if not task.done(): + task.cancel() + await asyncio.gather(beater, worker, return_exceptions=True) + ''' + # The literals above are indented inside this workflow script; normalize + # only that script indentation, not repository source indentation. + old = old.replace(" ", "") + new = new.replace(" ", "") + if text.count(old) != 1: + raise SystemExit("expected heartbeat cleanup block exactly once") + path.write_text(text.replace(old, new, 1)) + PY + git diff --check + git diff -- backend/app/global_ask_queue.py + + - name: Focused GREEN + run: | + set -euo pipefail + uv run --frozen python -m pytest -q \ + tests/test_global_ask_claim_cancellation.py \ + tests/test_global_ask_queue.py + + - name: Full PostgreSQL-backed GREEN + run: uv run --frozen python -m pytest -q + + - name: Remove temporary repair workflow and promote non-force + env: + GH_TOKEN: ${{ github.token }} + run: | + set -euo pipefail + live_head="$(gh api repos/${GITHUB_REPOSITORY}/pulls/979 --jq .head.sha)" + test "$live_head" = "$GITHUB_SHA" + git rm .github/workflows/automation-979-cancellation-repair.yml + git add backend/app/global_ask_queue.py + git diff --cached --check + git config user.name "github-actions[bot]" + git config user.email "41898282+github-actions[bot]@users.noreply.github.com" + git commit -m "fix(ask): cancel active operation with claim owner" + git push origin HEAD:"$PRODUCT_BRANCH" From 3dbba6b15733c6e517d70461446eb1049eba57bf Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 7 Sep 2026 23:45:03 +0900 Subject: [PATCH 06/23] fix(ask): cancel the inner Ask operation with its owner Owner-task cancellation left compute_global_ask_answer running as a detached child of the claim-heartbeat wrapper. Cancel and await that worker in finally, and remove the one-shot repair workflow whose RED now lives in the product test suite. --- .../automation-979-cancellation-repair.yml | 154 ------------------ backend/app/global_ask_queue.py | 5 +- docs/product-technical-gap-baseline.md | 7 +- 3 files changed, 6 insertions(+), 160 deletions(-) delete mode 100644 .github/workflows/automation-979-cancellation-repair.yml diff --git a/.github/workflows/automation-979-cancellation-repair.yml b/.github/workflows/automation-979-cancellation-repair.yml deleted file mode 100644 index 2fd7fb923..000000000 --- a/.github/workflows/automation-979-cancellation-repair.yml +++ /dev/null @@ -1,154 +0,0 @@ -name: Automation 979 cancellation repair - -on: - push: - branches: - - feat/ask-ownership-fenced-liveness-20260907 - -permissions: - contents: write - pull-requests: read - -concurrency: - group: automation-979-cancellation-repair - cancel-in-progress: false - -jobs: - repair: - if: github.actor != 'github-actions[bot]' - runs-on: ubuntu-24.04 - services: - postgres: - image: postgres:16-alpine@sha256:57c72fd2a128e416c7fcc499958864df5301e940bca0a56f58fddf30ffc07777 - env: - POSTGRES_PASSWORD: postgres - ports: - - 5432:5432 - options: >- - --health-cmd "pg_isready -U postgres" - --health-interval 10s - --health-timeout 5s - --health-retries 5 - env: - LINEAGEWEAVE_TEST_POSTGRES_ADMIN_DSN: postgresql://postgres:postgres@localhost:5432/postgres - EXPECTED_PRODUCT_PARENT: ee4430fa20f5cfe9c0d623061d92bc662c3353b8 - PRODUCT_BRANCH: feat/ask-ownership-fenced-liveness-20260907 - steps: - - name: Checkout exact automation head - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # actions/checkout@v7 - with: - fetch-depth: 0 - - - name: Verify authority before repair - env: - GH_TOKEN: ${{ github.token }} - run: | - set -euo pipefail - test "$(git rev-parse HEAD^)" = "$EXPECTED_PRODUCT_PARENT" - live_head="$(gh api repos/${GITHUB_REPOSITORY}/pulls/979 --jq .head.sha)" - test "$live_head" = "$GITHUB_SHA" - test "$(git status --porcelain)" = "" - - - name: Set up Python - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # actions/setup-python@v6 - with: - python-version: "3.12" - - - name: Set up locked dependency manager - uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0 - with: - version: "0.11.28" - enable-cache: false - - - name: Select pinned Rust toolchain - run: | - rustup toolchain install 1.97.1 --profile minimal - rustup default 1.97.1 - - - name: Install committed dependency graph - run: uv sync --frozen --extra dev --extra backend - - - name: Reproduce external-cancellation RED - run: | - set -euo pipefail - set +e - uv run --frozen python -m pytest -q tests/test_global_ask_claim_cancellation.py > /tmp/red.log 2>&1 - red_status=$? - set -e - cat /tmp/red.log - test "$red_status" -ne 0 - grep -F "owner cancellation left compute_global_ask_answer detached" /tmp/red.log - - - name: Apply minimum causal cancellation repair - run: | - set -euo pipefail - python - <<'PY' - from pathlib import Path - - path = Path("backend/app/global_ask_queue.py") - text = path.read_text() - old = ''' beater = asyncio.create_task(beat()) - worker = asyncio.create_task(operation) - try: - await asyncio.wait({worker, beater}, return_when=asyncio.FIRST_COMPLETED) - if lost.is_set(): - worker.cancel() - await asyncio.gather(worker, return_exceptions=True) - raise _LostAskClaim() - return await worker - finally: - stop.set() - beater.cancel() - await asyncio.gather(beater, return_exceptions=True) - ''' - new = ''' beater = asyncio.create_task(beat()) - worker = asyncio.create_task(operation) - try: - await asyncio.wait({worker, beater}, return_when=asyncio.FIRST_COMPLETED) - if lost.is_set(): - worker.cancel() - await asyncio.gather(worker, return_exceptions=True) - raise _LostAskClaim() - return await worker - finally: - stop.set() - for task in (beater, worker): - if not task.done(): - task.cancel() - await asyncio.gather(beater, worker, return_exceptions=True) - ''' - # The literals above are indented inside this workflow script; normalize - # only that script indentation, not repository source indentation. - old = old.replace(" ", "") - new = new.replace(" ", "") - if text.count(old) != 1: - raise SystemExit("expected heartbeat cleanup block exactly once") - path.write_text(text.replace(old, new, 1)) - PY - git diff --check - git diff -- backend/app/global_ask_queue.py - - - name: Focused GREEN - run: | - set -euo pipefail - uv run --frozen python -m pytest -q \ - tests/test_global_ask_claim_cancellation.py \ - tests/test_global_ask_queue.py - - - name: Full PostgreSQL-backed GREEN - run: uv run --frozen python -m pytest -q - - - name: Remove temporary repair workflow and promote non-force - env: - GH_TOKEN: ${{ github.token }} - run: | - set -euo pipefail - live_head="$(gh api repos/${GITHUB_REPOSITORY}/pulls/979 --jq .head.sha)" - test "$live_head" = "$GITHUB_SHA" - git rm .github/workflows/automation-979-cancellation-repair.yml - git add backend/app/global_ask_queue.py - git diff --cached --check - git config user.name "github-actions[bot]" - git config user.email "41898282+github-actions[bot]@users.noreply.github.com" - git commit -m "fix(ask): cancel active operation with claim owner" - git push origin HEAD:"$PRODUCT_BRANCH" diff --git a/backend/app/global_ask_queue.py b/backend/app/global_ask_queue.py index 9c5b1744c..c67e7e579 100644 --- a/backend/app/global_ask_queue.py +++ b/backend/app/global_ask_queue.py @@ -163,14 +163,13 @@ async def beat() -> None: try: await asyncio.wait({worker, beater}, return_when=asyncio.FIRST_COMPLETED) if lost.is_set(): - worker.cancel() - await asyncio.gather(worker, return_exceptions=True) raise _LostAskClaim() return await worker finally: stop.set() + worker.cancel() beater.cancel() - await asyncio.gather(beater, return_exceptions=True) + await asyncio.gather(worker, beater, return_exceptions=True) async def enqueue_global_ask_job( diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 25f4f020d..14dadbd05 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -6,9 +6,10 @@ > owner lost the generation and is not a buyer-visible failure. Orphan > recovery can still reclaim by age. Live workers renew ``updated_at`` > on the recovery interval so a current owner is not reclaimed by -> elapsed time. The 600 s worker deadline remains until that heartbeat -> evidence is independently approved. Draft; not protected-main or -> independently approved evidence. +> elapsed time. Cancelling the owner task also cancels the inner Ask +> operation so compute cannot continue detached. The 600 s worker +> deadline remains until that heartbeat evidence is independently +> approved. Draft; not protected-main or independently approved evidence. > > Exact-head loop overlay: 2026-08-29 13:20 KST. Protected `main` is > `fc13acaa20adca11968238e398d4aafcf62b6cee` (v2.23.0 leftover-map From dc061470b604ca317e57edc58af11b9f7299390a Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 8 Sep 2026 00:46:01 +0900 Subject: [PATCH 07/23] fix(ask): do not invent a 570-second Ask socket hang-up Omitted ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS now means no LineageWeave elapsed socket limit, matching post-chat's null transport default. An explicit finite value still has to stay below the 600 s job deadline, which remains until the heartbeat is independently approved. --- backend/app/config.py | 25 +++++++++++++------------ backend/app/main.py | 3 +-- backend/tests/test_config.py | 25 ++++++++++++++++++++++++- docs/product-technical-gap-baseline.md | 8 +++++--- 4 files changed, 43 insertions(+), 18 deletions(-) diff --git a/backend/app/config.py b/backend/app/config.py index 0fea9a591..96964bd20 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -51,11 +51,11 @@ class Settings: frontend_origins: list[str] orchestrator_base_url: str orchestrator_api_key: str - # Socket timeout for one Ask answer round-trip. Must stay below the Ask - # worker's job deadline so the client, not the job reaper, ends a slow - # call — hanging up earlier discards an answer the orchestrator has - # already paid to generate (observed live as a BrokenPipe on its side). - orchestrator_answer_timeout_seconds: float + # Optional socket timeout for one Ask answer round-trip. Omitted/blank + # means no LineageWeave elapsed socket limit (provider/orchestrator + # transport policy owns hang-up). An explicit finite value must stay + # below the Ask worker's job deadline. + orchestrator_answer_timeout_seconds: float | None valkey_url: str searxng_base_url: str tepp_transport_url: str @@ -82,14 +82,15 @@ def keycloak_jwks_uri(self) -> str: return f"{self.keycloak_base_url}/realms/{self.keycloak_realm}/protocol/openid-connect/certs" -def _validated_answer_timeout(raw: str) -> float: - """Parse the Ask answer timeout and hold it under the job deadline. +def _validated_answer_timeout(raw: str | None) -> float | None: + """Parse an optional Ask answer timeout under the job deadline. - The client must hang up before the worker's deadline reaper so a slow - answer settles as a clean client timeout, never a reaped job — values - at or above the deadline (or non-finite/non-positive ones) silently - break that ordering, so they are configuration errors. + Blank or omitted leaves no LineageWeave elapsed socket limit. An + explicit finite value must stay below the worker deadline so a + configured hang-up cannot outlive the job reaper. """ + if raw is None or not str(raw).strip(): + return None try: value = float(raw) except ValueError as exc: @@ -201,7 +202,7 @@ def load_settings() -> Settings: orchestrator_base_url=os.environ.get("ORCHESTRATOR_BASE_URL", ""), orchestrator_api_key=os.environ.get("ORCHESTRATOR_API_KEY", ""), orchestrator_answer_timeout_seconds=_validated_answer_timeout( - os.environ.get("ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS", "570") + os.environ.get("ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS") ), valkey_url=os.environ.get("VALKEY_URL", "redis://localhost:16379/0"), searxng_base_url=os.environ.get("SEARXNG_BASE_URL", ""), diff --git a/backend/app/main.py b/backend/app/main.py index 56b519ef4..ee2075739 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -323,8 +323,7 @@ async def lifespan(app: FastAPI): app.state.post_content_worker = content_worker # Late-bound lambda so tests that monkeypatch _post_chat_client reach # the worker too (the name resolves in module globals at call time). - # This worker still has an explicit answer socket limit; per-post - # chat retains the default null transport timeout. + # Ask and per-post chat both default to a null transport timeout. global_ask_worker = asyncio.create_task( run_global_ask_worker( valkey, diff --git a/backend/tests/test_config.py b/backend/tests/test_config.py index f37d89f32..5f8ff7783 100644 --- a/backend/tests/test_config.py +++ b/backend/tests/test_config.py @@ -2,7 +2,30 @@ from __future__ import annotations -from backend.app.config import load_settings +import pytest + +from backend.app.config import GLOBAL_ASK_JOB_DEADLINE_SECONDS, load_settings + + +def test_ask_answer_timeout_defaults_to_no_elapsed_socket_limit(monkeypatch) -> None: + """Omitted Ask socket timeout is not a hidden 570-second hang-up.""" + monkeypatch.delenv("ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS", raising=False) + assert load_settings().orchestrator_answer_timeout_seconds is None + + +def test_ask_answer_timeout_keeps_an_explicit_finite_value_under_the_deadline( + monkeypatch, +) -> None: + """An operator-set Ask socket timeout must stay below the job deadline.""" + monkeypatch.setenv("ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS", "570") + assert load_settings().orchestrator_answer_timeout_seconds == 570 + + monkeypatch.setenv( + "ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS", + str(GLOBAL_ASK_JOB_DEADLINE_SECONDS), + ) + with pytest.raises(ValueError, match="less than"): + load_settings() def test_frontend_origins_are_parsed_from_comma_separated_env(monkeypatch) -> None: diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 14dadbd05..d7139615c 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -7,9 +7,11 @@ > recovery can still reclaim by age. Live workers renew ``updated_at`` > on the recovery interval so a current owner is not reclaimed by > elapsed time. Cancelling the owner task also cancels the inner Ask -> operation so compute cannot continue detached. The 600 s worker -> deadline remains until that heartbeat evidence is independently -> approved. Draft; not protected-main or independently approved evidence. +> operation so compute cannot continue detached. Ask HTTP no longer +> invents a 570 s socket hang-up when the operator omits a timeout; the +> 600 s worker deadline remains until that heartbeat evidence is +> independently approved. Not protected-main or independently approved +> evidence. > > Exact-head loop overlay: 2026-08-29 13:20 KST. Protected `main` is > `fc13acaa20adca11968238e398d4aafcf62b6cee` (v2.23.0 leftover-map From 4a33f0e12804c36c163b831d4a214eaabc33af9f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 8 Sep 2026 01:07:28 +0900 Subject: [PATCH 08/23] test(ask): keep live heartbeat work past elapsed deadline --- tests/test_global_ask_elapsed_deadline.py | 82 +++++++++++++++++++++++ 1 file changed, 82 insertions(+) create mode 100644 tests/test_global_ask_elapsed_deadline.py diff --git a/tests/test_global_ask_elapsed_deadline.py b/tests/test_global_ask_elapsed_deadline.py new file mode 100644 index 000000000..79061976f --- /dev/null +++ b/tests/test_global_ask_elapsed_deadline.py @@ -0,0 +1,82 @@ +"""Regression for Global Ask liveness versus elapsed-time cancellation.""" + +from __future__ import annotations + +import asyncio +from contextlib import asynccontextmanager +from datetime import UTC, datetime, timedelta + +from backend.app import global_ask_queue + + +_CLAIMED_AT = datetime(2026, 1, 1, 9, 0, tzinfo=UTC) + + +class _AvailableClient: + available = True + + +class _Connection: + def __init__(self) -> None: + self.generation = _CLAIMED_AT + self.executed: list[tuple[str, tuple[object, ...]]] = [] + + async def fetchrow(self, query: str, *_args: object): + if "job_status_code = $3" in query: + return { + "requesting_account_id": "00000000-0000-0000-0000-000000000001", + "question_text": "Continue the live Ask operation", + "verify_external_requested": False, + "knowledge_cutoff": None, + "updated_at": self.generation, + } + if "returning updated_at" in query: + self.generation += timedelta(milliseconds=5) + return {"updated_at": self.generation} + raise AssertionError(query) + + async def execute(self, query: str, *args: object) -> str: + self.executed.append((query, args)) + return "UPDATE 1" + + +class _Pool: + def __init__(self, connection: _Connection) -> None: + self.connection = connection + + @asynccontextmanager + async def acquire(self): + yield self.connection + + +def test_live_heartbeat_operation_is_not_cancelled_by_elapsed_time(monkeypatch) -> None: + """A renewing owner may outlive the former hard worker deadline.""" + connection = _Connection() + pool = _Pool(connection) + + async def _load_visibility(_conn, _job_id, _account_id): + return {"corp-1"}, set(), False, True + + async def _long_but_live_answer(*_args, **_kwargs): + await asyncio.sleep(0.03) + return {"answer_text": "completed by live owner"} + + monkeypatch.setattr(global_ask_queue, "JOB_DEADLINE_SECONDS", 0.01) + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.005) + monkeypatch.setattr(global_ask_queue, "load_job_visibility", _load_visibility) + monkeypatch.setattr( + global_ask_queue, "compute_global_ask_answer", _long_but_live_answer + ) + + asyncio.run( + global_ask_queue.process_global_ask_job( + pool, + job_id="job-live", + chat_factory=_AvailableClient, + ) + ) + + settle_query, settle_args = connection.executed[-1] + assert "answer_payload" in settle_query + assert "failure_detail" not in settle_query + assert settle_args[3:] == (global_ask_queue.RUNNING, connection.generation) From f1a71aeed44c5953fab6a02cdbb3dc8a3b24cda8 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 8 Sep 2026 01:47:16 +0900 Subject: [PATCH 09/23] fix(ask): keep live heartbeat work past elapsed deadline The 600 s asyncio timeout still cancelled a renewing Ask owner. Compute now runs under claim-generation liveness only. ADR 0370 records that Proposed policy. Orphan recovery uses three missed heartbeats. Provider TimeoutError stays unavailable, not a worker deadline. PostgreSQL-backed race evidence remains a follow-up. --- backend/app/global_ask_queue.py | 64 ++++++++----------- .../adr/0370-ask-claim-generation-liveness.md | 47 ++++++++++++++ docs/product-technical-gap-baseline.md | 6 +- tests/test_global_ask_queue.py | 17 ++--- 4 files changed, 81 insertions(+), 53 deletions(-) create mode 100644 docs/adr/0370-ask-claim-generation-liveness.md diff --git a/backend/app/global_ask_queue.py b/backend/app/global_ask_queue.py index c67e7e579..c06c6517b 100644 --- a/backend/app/global_ask_queue.py +++ b/backend/app/global_ask_queue.py @@ -76,17 +76,16 @@ # trimmed stream) and are republished by the worker's recovery sweep. _REPUBLISH_AFTER_SECONDS = 60 _RECOVERY_INTERVAL_SECONDS = 30.0 -# Hard ceiling on one job's answer computation. Without it a hung -# orchestrator round-trip kept a job `running` indefinitely (observed: -# 17+ minutes) and, before concurrent processing, stalled every job -# behind it. Shared through config so the client-timeout validation and -# this reaper can never disagree. +# Live workers renew the claim generation on this interval so age-based +# orphan recovery cannot reclaim a job that is still owned. +_CLAIM_HEARTBEAT_SECONDS = _RECOVERY_INTERVAL_SECONDS +# Optional explicit Ask HTTP hang-up still has to stay below this bound. +# Live compute is no longer cancelled when this many seconds elapse. JOB_DEADLINE_SECONDS = GLOBAL_ASK_JOB_DEADLINE_SECONDS -# A `running` row older than this is an orphan: a live worker's deadline -# settles every job within JOB_DEADLINE_SECONDS, so one sweep interval of -# slack past that is enough — recovering sooner shortens how long a -# crashed worker's job stays invisible to a polling reader. -_ORPHAN_RUNNING_AFTER_SECONDS = JOB_DEADLINE_SECONDS + 60 +# A `running` row whose claim generation has not been renewed for this +# many seconds is an orphan. Live workers heartbeat more often, so age +# alone does not reclaim a current owner. +_ORPHAN_RUNNING_AFTER_SECONDS = 3 * _CLAIM_HEARTBEAT_SECONDS # Wake-up stream cap, mirroring the post-content stream: the durable rows # are the source of truth, so trimming old wake-ups loses nothing. _STREAM_MAX_LENGTH = 1000 @@ -94,9 +93,6 @@ # LLM round-trips, so serial consumption would head-of-line block every # later question behind the slowest one. _WORKER_CONCURRENCY = 4 -# Live workers renew the claim generation on this interval so age-based -# orphan recovery cannot reclaim a job that is still owned. -_CLAIM_HEARTBEAT_SECONDS = _RECOVERY_INTERVAL_SECONDS _logger = logging.getLogger(__name__) @@ -610,7 +606,6 @@ async def process_global_ask_job( if row is None: return lease = [row["updated_at"]] - answer_timeout: asyncio.Timeout | None = None try: async with pool.acquire() as conn: ( @@ -628,25 +623,24 @@ async def process_global_ask_job( raise _SafeJobError( "Ask Agent is unavailable: set ORCHESTRATOR_BASE_URL / ORCHESTRATOR_API_KEY" ) - async with asyncio.timeout(JOB_DEADLINE_SECONDS) as answer_timeout: - payload = await _run_with_ask_claim_heartbeat( + payload = await _run_with_ask_claim_heartbeat( + pool, + job_id, + lease, + compute_global_ask_answer( pool, - job_id, - lease, - compute_global_ask_answer( - pool, - question_text=str(row["question_text"]), - corporate_entity_ids=entity_ids, - process_unit_ids=process_unit_ids, - process_scope_limited=process_scope_limited, - chat_client=chat_client, - embedding_client=embedding_factory(), - semantic_query_client=semantic_query_factory(), - verify_external=bool(row["verify_external_requested"]), - claim_verification_client=claim_verification_factory(), - knowledge_cutoff=row["knowledge_cutoff"], - ), - ) + question_text=str(row["question_text"]), + corporate_entity_ids=entity_ids, + process_unit_ids=process_unit_ids, + process_scope_limited=process_scope_limited, + chat_client=chat_client, + embedding_client=embedding_factory(), + semantic_query_client=semantic_query_factory(), + verify_external=bool(row["verify_external_requested"]), + claim_verification_client=claim_verification_factory(), + knowledge_cutoff=row["knowledge_cutoff"], + ), + ) except _LostAskClaim: return except asyncio.CancelledError: @@ -666,12 +660,6 @@ async def process_global_ask_job( # Raised locally with a pre-authored, safe message (permission # state / missing config) — never a provider-boundary leak. detail = str(exc) - elif ( - isinstance(exc, asyncio.TimeoutError) - and answer_timeout is not None - and answer_timeout.expired() - ): - detail = f"job exceeded the {JOB_DEADLINE_SECONDS}s deadline" else: # Provider responses/exceptions can carry credentials, gateway # diagnostics, or model output (ADR 0123): never persist the diff --git a/docs/adr/0370-ask-claim-generation-liveness.md b/docs/adr/0370-ask-claim-generation-liveness.md new file mode 100644 index 000000000..46bef068c --- /dev/null +++ b/docs/adr/0370-ask-claim-generation-liveness.md @@ -0,0 +1,47 @@ +# ADR 0370 — Global Ask claim-generation liveness + +**Decision status:** Proposed +**Date:** 2026-09-08 + +Amends the Global Ask worker settlement path. Independent of leftover-map +ADRs 0272+ on other stacks and of the versioned translation ledger +([ADR 0362](0362-versioned-ui-translation-ledger.md)). + +## Context + +Issue #975: a live Ask job could be terminated or reclaimed from elapsed +wall time (600 s compute deadline, 570 s invented socket hang-up) while +a provider `TimeoutError` and a worker deadline meant different things. +Orphan recovery flipped `running` → `queued` by `updated_at` age, and +the original worker could still settle by job id alone. + +## Decision + +- Claim `queued` → `running` returns a generation (`updated_at`). +- Settlement is compare-and-set on that generation; PostgreSQL + `UPDATE 0` is an unapplied settle, not a buyer-visible failure. +- While computing, the owner renews `updated_at` on the recovery + interval. A failed renew aborts without settling as failed. +- Cancelling the owner task cancels the inner compute task. +- LineageWeave does not invent an Ask socket hang-up when + `ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS` is omitted. An explicit finite + value must stay below 600 s. +- Live compute is not cancelled when 600 s elapse. Age-based orphan + recovery uses three missed heartbeats, not the old 660 s reaper. +- Provider `TimeoutError` stays an unavailable Ask failure. + +## Consequences + +Positive: a renewing owner can outlive the former hard deadline; a +reclaimed job cannot be overwritten by the previous owner. + +Negative: crashed workers wait three heartbeat intervals to reclaim. +PostgreSQL-backed claim/settlement evidence for the race remains a +follow-up; current REDs use deterministic fake pools plus exact SQL +contracts. + +## Alternatives considered + +Keep the 600 s `asyncio.timeout` around compute: rejected because it +cancels a live heartbeat owner. The elapsed-deadline RED proves that +path. diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index d7139615c..40297827c 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -9,9 +9,9 @@ > elapsed time. Cancelling the owner task also cancels the inner Ask > operation so compute cannot continue detached. Ask HTTP no longer > invents a 570 s socket hang-up when the operator omits a timeout; the -> 600 s worker deadline remains until that heartbeat evidence is -> independently approved. Not protected-main or independently approved -> evidence. +> 600 s compute hang-up is removed on this branch (ADR 0370 Proposed): +> live heartbeat owners are not cancelled by elapsed time. Not +> protected-main or independently approved evidence. > > Exact-head loop overlay: 2026-08-29 13:20 KST. Protected `main` is > `fc13acaa20adca11968238e398d4aafcf62b6cee` (v2.23.0 leftover-map diff --git a/tests/test_global_ask_queue.py b/tests/test_global_ask_queue.py index 4a82b3ca5..d0aadc393 100644 --- a/tests/test_global_ask_queue.py +++ b/tests/test_global_ask_queue.py @@ -442,7 +442,7 @@ async def _fake_compute_global_ask_answer(*_args, **_kwargs): assert settle_args[3:] == (global_ask_queue.RUNNING, _CLAIMED_AT) -@pytest.mark.parametrize("timeout_source", ["provider", "worker", "shutdown"]) +@pytest.mark.parametrize("timeout_source", ["provider", "shutdown"]) def test_timeout_detail_identifies_only_an_expired_worker_deadline( monkeypatch, timeout_source, ) -> None: @@ -456,12 +456,8 @@ async def _fake_load_job_visibility(_conn, _job_id, _account_id): async def _fake_compute_global_ask_answer(*_args, **_kwargs): if timeout_source == "shutdown": raise asyncio.CancelledError() - if timeout_source == "worker": - await asyncio.Event().wait() raise asyncio.TimeoutError("synthetic private upstream detail") - if timeout_source == "worker": - monkeypatch.setattr(global_ask_queue, "JOB_DEADLINE_SECONDS", 0) monkeypatch.setattr(global_ask_queue, "load_job_visibility", _fake_load_job_visibility) monkeypatch.setattr( global_ask_queue, "compute_global_ask_answer", _fake_compute_global_ask_answer @@ -484,13 +480,10 @@ async def _fake_compute_global_ask_answer(*_args, **_kwargs): ) _settle_query, settle_args = connection.executed[-1] - if timeout_source == "worker": - assert settle_args[2] == "job exceeded the 0s deadline" - else: - assert settle_args[2] == ( - "Ask Agent is unavailable: contextual-orchestrator returned " - "no complete evidence object" - ) + assert settle_args[2] == ( + "Ask Agent is unavailable: contextual-orchestrator returned " + "no complete evidence object" + ) assert settle_args[3:] == (global_ask_queue.RUNNING, _CLAIMED_AT) From fbe68078a325d760e82105b7175492176d29453d Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 8 Sep 2026 02:45:56 +0900 Subject: [PATCH 10/23] test(ask): prove stale owner settlement fails on real PostgreSQL Issue #975 requires the reclaim invariant against actual claim and settlement transactions, not only in-memory fakes. After 0001 and 0165 replay, the previous owner's compare-and-set is UPDATE 0 and the new owner settles. --- .../adr/0370-ask-claim-generation-liveness.md | 7 +- docs/product-technical-gap-baseline.md | 5 +- tests/test_schema.py | 100 ++++++++++++++++++ 3 files changed, 107 insertions(+), 5 deletions(-) diff --git a/docs/adr/0370-ask-claim-generation-liveness.md b/docs/adr/0370-ask-claim-generation-liveness.md index 46bef068c..2363619d3 100644 --- a/docs/adr/0370-ask-claim-generation-liveness.md +++ b/docs/adr/0370-ask-claim-generation-liveness.md @@ -36,9 +36,10 @@ Positive: a renewing owner can outlive the former hard deadline; a reclaimed job cannot be overwritten by the previous owner. Negative: crashed workers wait three heartbeat intervals to reclaim. -PostgreSQL-backed claim/settlement evidence for the race remains a -follow-up; current REDs use deterministic fake pools plus exact SQL -contracts. + +`tests/test_schema.py` proves the settlement compare-and-set against a +throwaway database that replayed the real `0001` and `0165` migrations. +In-memory queue tests remain for elapsed-time and cancellation contracts. ## Alternatives considered diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 40297827c..bc07657fd 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -10,8 +10,9 @@ > operation so compute cannot continue detached. Ask HTTP no longer > invents a 570 s socket hang-up when the operator omits a timeout; the > 600 s compute hang-up is removed on this branch (ADR 0370 Proposed): -> live heartbeat owners are not cancelled by elapsed time. Not -> protected-main or independently approved evidence. +> live heartbeat owners are not cancelled by elapsed time. Real +> PostgreSQL compare-and-set reclaim is covered in `test_schema.py`. +> Not protected-main or independently approved evidence. > > Exact-head loop overlay: 2026-08-29 13:20 KST. Protected `main` is > `fc13acaa20adca11968238e398d4aafcf62b6cee` (v2.23.0 leftover-map diff --git a/tests/test_schema.py b/tests/test_schema.py index 94876295f..b14ec94f9 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -325,6 +325,106 @@ def test_migration_applies_cleanly(schema_db) -> None: assert expected <= tables +def test_postgres_stale_ask_owner_cannot_settle_after_reclaim(schema_db) -> None: + """#975: real PostgreSQL compare-and-set blocks the previous Ask owner.""" + with schema_db.cursor() as cur: + cur.execute( + """ + insert into user_account (external_subject_id, display_name, email_address) + values ('ask-claim-generation', 'Ask claim generation', 'ask-claim@example.test') + returning user_account_id + """ + ) + account_id = cur.fetchone()[0] + cur.execute( + """ + insert into global_ask_job + (requesting_account_id, question_text, job_status_code) + values (%s, 'reclaim settlement', 'queued') + returning global_ask_job_id + """, + (account_id,), + ) + job_id = cur.fetchone()[0] + schema_db.commit() + + parsed_admin_dsn = urlsplit(_ADMIN_DSN) + db_dsn = urlunsplit( + parsed_admin_dsn._replace(path=f"/{schema_db.get_dsn_parameters()['dbname']}") + ) + + async def race() -> None: + owner = await asyncpg.connect(db_dsn) + reclaim = await asyncpg.connect(db_dsn) + try: + claimed = await owner.fetchrow( + """ + update global_ask_job set job_status_code = $2, updated_at = now() + where global_ask_job_id = $1 and job_status_code = $3 + returning updated_at + """, + job_id, + "running", + "queued", + ) + assert claimed is not None + await reclaim.execute( + """ + update global_ask_job set job_status_code = $1, updated_at = now() + where global_ask_job_id = $2 and job_status_code = $3 + """, + "queued", + job_id, + "running", + ) + reclaimed = await reclaim.fetchrow( + """ + update global_ask_job set job_status_code = $2, updated_at = now() + where global_ask_job_id = $1 and job_status_code = $3 + returning updated_at + """, + job_id, + "running", + "queued", + ) + assert reclaimed is not None + stale = await owner.execute( + """ + update global_ask_job set job_status_code = $2, + answer_payload = $3::jsonb, updated_at = now() + where global_ask_job_id = $1 + and job_status_code = $4 + and updated_at = $5 + """, + job_id, + "succeeded", + '{"answer_text":"stale-owner"}', + "running", + claimed["updated_at"], + ) + live = await reclaim.execute( + """ + update global_ask_job set job_status_code = $2, + answer_payload = $3::jsonb, updated_at = now() + where global_ask_job_id = $1 + and job_status_code = $4 + and updated_at = $5 + """, + job_id, + "succeeded", + '{"answer_text":"live-owner"}', + "running", + reclaimed["updated_at"], + ) + assert stale == "UPDATE 0" + assert live == "UPDATE 1" + finally: + await owner.close() + await reclaim.close() + + asyncio.run(race()) + + def test_occupational_catalog_metadata_columns_exist(schema_db) -> None: """The real schema preserves catalog descriptions and release integrity.""" with schema_db.cursor() as cur: From 70cdd9c4f5dd673e4376130dbfa8585218b029cc Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 8 Sep 2026 03:21:42 +0900 Subject: [PATCH 11/23] test(ask): decouple transport timeout from removed worker deadline --- tests/test_global_ask_transport_timeout.py | 26 ++++++++++++++++++++++ 1 file changed, 26 insertions(+) create mode 100644 tests/test_global_ask_transport_timeout.py diff --git a/tests/test_global_ask_transport_timeout.py b/tests/test_global_ask_transport_timeout.py new file mode 100644 index 000000000..c8653e976 --- /dev/null +++ b/tests/test_global_ask_transport_timeout.py @@ -0,0 +1,26 @@ +"""Regression contracts for the optional Global Ask transport timeout.""" + +from __future__ import annotations + +import pytest + +from backend.app.config import _validated_answer_timeout + + +def test_explicit_transport_timeout_is_not_coupled_to_removed_worker_deadline() -> None: + """A finite positive transport timeout is independent of worker liveness.""" + assert _validated_answer_timeout("900") == 900.0 + + +def test_explicit_transport_timeout_rejects_non_positive_or_non_finite_values() -> None: + """Explicit transport limits must be finite and strictly positive.""" + for raw in ("0", "-1", "nan", "inf", "-inf"): + with pytest.raises(ValueError): + _validated_answer_timeout(raw) + + +def test_omitted_or_blank_transport_timeout_remains_null() -> None: + """No configured timeout means no LineageWeave elapsed transport limit.""" + assert _validated_answer_timeout(None) is None + assert _validated_answer_timeout("") is None + assert _validated_answer_timeout(" ") is None From 69968682b11cd86c8cfaba11620c7a85b270a28c Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 8 Sep 2026 03:26:54 +0900 Subject: [PATCH 12/23] fix(ask): separate transport timeout from worker liveness --- backend/app/config.py | 23 ++++++++--------------- 1 file changed, 8 insertions(+), 15 deletions(-) diff --git a/backend/app/config.py b/backend/app/config.py index 96964bd20..b1199e7dc 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -8,11 +8,6 @@ import os from dataclasses import dataclass, field -# Hard ceiling on one Global Ask job's answer computation, shared with the -# worker in global_ask_queue.py so config validation and execution can never -# disagree about the bound. -GLOBAL_ASK_JOB_DEADLINE_SECONDS = 600 - @dataclass(frozen=True) class Settings: @@ -52,9 +47,8 @@ class Settings: orchestrator_base_url: str orchestrator_api_key: str # Optional socket timeout for one Ask answer round-trip. Omitted/blank - # means no LineageWeave elapsed socket limit (provider/orchestrator - # transport policy owns hang-up). An explicit finite value must stay - # below the Ask worker's job deadline. + # means no LineageWeave elapsed socket limit. Explicit values are + # deployment transport policy and remain independent of worker liveness. orchestrator_answer_timeout_seconds: float | None valkey_url: str searxng_base_url: str @@ -83,11 +77,11 @@ def keycloak_jwks_uri(self) -> str: def _validated_answer_timeout(raw: str | None) -> float | None: - """Parse an optional Ask answer timeout under the job deadline. + """Parse an optional finite-positive Ask transport timeout. - Blank or omitted leaves no LineageWeave elapsed socket limit. An - explicit finite value must stay below the worker deadline so a - configured hang-up cannot outlive the job reaper. + Blank or omitted leaves no LineageWeave elapsed socket limit. Worker + liveness is owned by claim heartbeats and generation fencing, not by + this optional transport policy. """ if raw is None or not str(raw).strip(): return None @@ -97,10 +91,9 @@ def _validated_answer_timeout(raw: str | None) -> float | None: raise ValueError( "ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS must be a number" ) from exc - if not math.isfinite(value) or not 0 < value < GLOBAL_ASK_JOB_DEADLINE_SECONDS: + if not math.isfinite(value) or value <= 0: raise ValueError( - "ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS must be a finite number greater" - f" than 0 and less than {GLOBAL_ASK_JOB_DEADLINE_SECONDS}" + "ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS must be a finite number greater than 0" ) return value From 5d0272fa647f631e116f7ed4a8311272174423cb Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 8 Sep 2026 03:28:54 +0900 Subject: [PATCH 13/23] fix(ask): keep removed deadline import non-operative --- backend/app/config.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/backend/app/config.py b/backend/app/config.py index b1199e7dc..5d0dbe5d2 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -8,6 +8,11 @@ import os from dataclasses import dataclass, field +# Compatibility sentinel for the stacked worker module while its obsolete +# deadline import is removed. It is deliberately non-operative: Global Ask +# liveness is claim-heartbeat/generation based and has no elapsed worker cap. +GLOBAL_ASK_JOB_DEADLINE_SECONDS = None + @dataclass(frozen=True) class Settings: From 4cf398f8306f6dc12984a5e70792303a83b6228e Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 8 Sep 2026 03:29:11 +0900 Subject: [PATCH 14/23] test(ask): stop depending on removed worker deadline symbol --- tests/test_global_ask_elapsed_deadline.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tests/test_global_ask_elapsed_deadline.py b/tests/test_global_ask_elapsed_deadline.py index 79061976f..ba826cca4 100644 --- a/tests/test_global_ask_elapsed_deadline.py +++ b/tests/test_global_ask_elapsed_deadline.py @@ -50,7 +50,7 @@ async def acquire(self): def test_live_heartbeat_operation_is_not_cancelled_by_elapsed_time(monkeypatch) -> None: - """A renewing owner may outlive the former hard worker deadline.""" + """A renewing owner remains live until completion, cancellation, or claim loss.""" connection = _Connection() pool = _Pool(connection) @@ -61,7 +61,6 @@ async def _long_but_live_answer(*_args, **_kwargs): await asyncio.sleep(0.03) return {"answer_text": "completed by live owner"} - monkeypatch.setattr(global_ask_queue, "JOB_DEADLINE_SECONDS", 0.01) monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.005) monkeypatch.setattr(global_ask_queue, "load_job_visibility", _load_visibility) monkeypatch.setattr( From 7d71a18d46314b411070472e1fc3a60384162178 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 8 Sep 2026 03:37:51 +0900 Subject: [PATCH 15/23] fix(ask): remove obsolete elapsed worker deadline --- backend/app/global_ask_queue.py | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/backend/app/global_ask_queue.py b/backend/app/global_ask_queue.py index c06c6517b..798730e9b 100644 --- a/backend/app/global_ask_queue.py +++ b/backend/app/global_ask_queue.py @@ -54,7 +54,6 @@ from lineageweave.semantic_query import NullSemanticQueryClient, SemanticQueryClient from lineageweave.temporal_expressions import resolve_korean_relative_time -from .config import GLOBAL_ASK_JOB_DEADLINE_SECONDS from .lineage_ingestion import lineage_graphs_for_posts from .operability import log_internal_fault, log_provider_unavailable from .post_chat_ingestion import ( @@ -79,9 +78,6 @@ # Live workers renew the claim generation on this interval so age-based # orphan recovery cannot reclaim a job that is still owned. _CLAIM_HEARTBEAT_SECONDS = _RECOVERY_INTERVAL_SECONDS -# Optional explicit Ask HTTP hang-up still has to stay below this bound. -# Live compute is no longer cancelled when this many seconds elapse. -JOB_DEADLINE_SECONDS = GLOBAL_ASK_JOB_DEADLINE_SECONDS # A `running` row whose claim generation has not been renewed for this # many seconds is an orphan. Live workers heartbeat more often, so age # alone does not reclaim a current owner. @@ -720,10 +716,9 @@ async def republish_queued_global_ask_jobs( A ``queued`` row older than the republish window lost its stream entry (crash or trim between insert and XADD). A ``running`` row - older than the orphan window belongs to a worker that died mid-job — - the per-job deadline guarantees a live worker settles sooner — so it - is flipped back to ``queued`` and re-woken for at-least-once - delivery. + whose claim generation stays stale beyond the orphan window belongs + to a worker that stopped heartbeating, so it is flipped back to + ``queued`` and re-woken for at-least-once delivery. """ async with pool.acquire() as conn: # Fully parameterized ($1..$3 with module constants); the rule From aecb873a91da3924a9ba763e1fd951332f794e42 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 8 Sep 2026 03:38:37 +0900 Subject: [PATCH 16/23] fix(ask): remove obsolete worker deadline sentinel --- backend/app/config.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/backend/app/config.py b/backend/app/config.py index 5d0dbe5d2..b1199e7dc 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -8,11 +8,6 @@ import os from dataclasses import dataclass, field -# Compatibility sentinel for the stacked worker module while its obsolete -# deadline import is removed. It is deliberately non-operative: Global Ask -# liveness is claim-heartbeat/generation based and has no elapsed worker cap. -GLOBAL_ASK_JOB_DEADLINE_SECONDS = None - @dataclass(frozen=True) class Settings: From 29d226529bdaf0b1fcc94e3228160c8935e703d4 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 8 Sep 2026 03:46:44 +0900 Subject: [PATCH 17/23] test(ask): drop leftover worker-deadline import from settings tests Collection on aecb873a imported GLOBAL_ASK_JOB_DEADLINE_SECONDS after the constant was removed. load_settings now accepts an explicit 900 s transport timeout and rejects non-positive values. ADR 0370 Proposed no longer bounds that timeout below 600 s. --- backend/tests/test_config.py | 22 +++++++++++-------- .../adr/0370-ask-claim-generation-liveness.md | 5 +++-- docs/product-technical-gap-baseline.md | 21 +++++++----------- 3 files changed, 24 insertions(+), 24 deletions(-) diff --git a/backend/tests/test_config.py b/backend/tests/test_config.py index 5f8ff7783..0469266cc 100644 --- a/backend/tests/test_config.py +++ b/backend/tests/test_config.py @@ -4,7 +4,7 @@ import pytest -from backend.app.config import GLOBAL_ASK_JOB_DEADLINE_SECONDS, load_settings +from backend.app.config import load_settings def test_ask_answer_timeout_defaults_to_no_elapsed_socket_limit(monkeypatch) -> None: @@ -13,19 +13,23 @@ def test_ask_answer_timeout_defaults_to_no_elapsed_socket_limit(monkeypatch) -> assert load_settings().orchestrator_answer_timeout_seconds is None -def test_ask_answer_timeout_keeps_an_explicit_finite_value_under_the_deadline( +def test_ask_answer_timeout_accepts_explicit_finite_values_without_worker_deadline( monkeypatch, ) -> None: - """An operator-set Ask socket timeout must stay below the job deadline.""" + """An operator-set Ask socket timeout is independent of worker liveness.""" monkeypatch.setenv("ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS", "570") assert load_settings().orchestrator_answer_timeout_seconds == 570 - monkeypatch.setenv( - "ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS", - str(GLOBAL_ASK_JOB_DEADLINE_SECONDS), - ) - with pytest.raises(ValueError, match="less than"): - load_settings() + monkeypatch.setenv("ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS", "900") + assert load_settings().orchestrator_answer_timeout_seconds == 900 + + +def test_ask_answer_timeout_rejects_non_positive_explicit_values(monkeypatch) -> None: + """Zero, negative, and non-finite Ask socket timeouts remain configuration errors.""" + for raw in ("0", "-1", "nan", "inf"): + monkeypatch.setenv("ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS", raw) + with pytest.raises(ValueError, match="finite number greater than 0"): + load_settings() def test_frontend_origins_are_parsed_from_comma_separated_env(monkeypatch) -> None: diff --git a/docs/adr/0370-ask-claim-generation-liveness.md b/docs/adr/0370-ask-claim-generation-liveness.md index 2363619d3..f2c33c8d8 100644 --- a/docs/adr/0370-ask-claim-generation-liveness.md +++ b/docs/adr/0370-ask-claim-generation-liveness.md @@ -24,8 +24,9 @@ the original worker could still settle by job id alone. interval. A failed renew aborts without settling as failed. - Cancelling the owner task cancels the inner compute task. - LineageWeave does not invent an Ask socket hang-up when - `ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS` is omitted. An explicit finite - value must stay below 600 s. + `ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS` is omitted or blank. An + explicit value must be finite and strictly positive; it is not + bounded by the removed 600 s worker deadline. - Live compute is not cancelled when 600 s elapse. Age-based orphan recovery uses three missed heartbeats, not the old 660 s reaper. - Provider `TimeoutError` stays an unavailable Ask failure. diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index bc07657fd..02fd125b0 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -1,18 +1,13 @@ # Product & Technical Gap Baseline -> Ask ownership-fence overlay: 2026-09-07 KST. Issue #975 stacks onto -> #974 so Global Ask settlement is compare-and-set on the claim -> generation (`updated_at`). A PostgreSQL `UPDATE 0` means the previous -> owner lost the generation and is not a buyer-visible failure. Orphan -> recovery can still reclaim by age. Live workers renew ``updated_at`` -> on the recovery interval so a current owner is not reclaimed by -> elapsed time. Cancelling the owner task also cancels the inner Ask -> operation so compute cannot continue detached. Ask HTTP no longer -> invents a 570 s socket hang-up when the operator omits a timeout; the -> 600 s compute hang-up is removed on this branch (ADR 0370 Proposed): -> live heartbeat owners are not cancelled by elapsed time. Real -> PostgreSQL compare-and-set reclaim is covered in `test_schema.py`. -> Not protected-main or independently approved evidence. +> Ask ownership-fence overlay: 2026-09-08 KST. Issue #975 / #979 exact +> head `aecb873a` removed the 600 s worker deadline, then Tests run +> `34152472692` failed at collection: `backend/tests/test_config.py` +> still imported `GLOBAL_ASK_JOB_DEADLINE_SECONDS`. That import is gone; +> `load_settings()` now accepts an explicit 900 s transport timeout and +> still rejects zero, negative, and non-finite values. ADR 0370 +> Proposed no longer requires an explicit timeout below 600 s. Frontend +> on that run was GREEN. Not protected-main or independently approved. > > Exact-head loop overlay: 2026-08-29 13:20 KST. Protected `main` is > `fc13acaa20adca11968238e398d4aafcf62b6cee` (v2.23.0 leftover-map From 262d496a91d867cb314f00d83863e94d78f50a87 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 8 Sep 2026 04:43:55 +0900 Subject: [PATCH 18/23] fix(ask): keep the claim heartbeat helper private Tests run 34153018910 failed the public-docstring gate because the nested Ask heartbeat coroutine was named beat. Rename it _beat so the helper is not a public production definition. --- backend/app/global_ask_queue.py | 4 ++-- docs/product-technical-gap-baseline.md | 14 +++++++------- 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/backend/app/global_ask_queue.py b/backend/app/global_ask_queue.py index 798730e9b..e25f4e0b1 100644 --- a/backend/app/global_ask_queue.py +++ b/backend/app/global_ask_queue.py @@ -138,7 +138,7 @@ async def _run_with_ask_claim_heartbeat( lost = asyncio.Event() stop = asyncio.Event() - async def beat() -> None: + async def _beat() -> None: while not stop.is_set() and not lost.is_set(): try: await asyncio.wait_for(stop.wait(), timeout=_CLAIM_HEARTBEAT_SECONDS) @@ -150,7 +150,7 @@ async def beat() -> None: return lease[0] = renewed - beater = asyncio.create_task(beat()) + beater = asyncio.create_task(_beat()) worker = asyncio.create_task(operation) try: await asyncio.wait({worker, beater}, return_when=asyncio.FIRST_COMPLETED) diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 02fd125b0..8ff70d131 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -1,13 +1,13 @@ # Product & Technical Gap Baseline > Ask ownership-fence overlay: 2026-09-08 KST. Issue #975 / #979 exact -> head `aecb873a` removed the 600 s worker deadline, then Tests run -> `34152472692` failed at collection: `backend/tests/test_config.py` -> still imported `GLOBAL_ASK_JOB_DEADLINE_SECONDS`. That import is gone; -> `load_settings()` now accepts an explicit 900 s transport timeout and -> still rejects zero, negative, and non-finite values. ADR 0370 -> Proposed no longer requires an explicit timeout below 600 s. Frontend -> on that run was GREEN. Not protected-main or independently approved. +> head `29d226529` cleared the settings-collection ImportError (Tests +> run `34152472692`). Follow-up Tests run `34153018910` then failed +> the public-docstring gate on a nested Ask heartbeat helper named +> without a leading underscore. That helper is now private. Frontend +> on that run was GREEN. Explicit 900 s transport timeouts remain +> valid. ADR 0370 Proposed. Not protected-main or independently +> approved. > > Exact-head loop overlay: 2026-08-29 13:20 KST. Protected `main` is > `fc13acaa20adca11968238e398d4aafcf62b6cee` (v2.23.0 leftover-map From 875e9364a9d10f0899c0077bdff2c94e58544576 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 8 Sep 2026 06:44:12 +0900 Subject: [PATCH 19/23] fix(ask): abort when the claim heartbeat dies first A heartbeat exception left compute running without renewals until the test timed out. Treat a finished heartbeat with a live worker as a lost claim so the owner is cancelled instead of settling. --- backend/app/global_ask_queue.py | 8 +++- .../adr/0370-ask-claim-generation-liveness.md | 2 + docs/product-technical-gap-baseline.md | 14 +++---- tests/test_global_ask_claim_cancellation.py | 39 +++++++++++++++++++ 4 files changed, 54 insertions(+), 9 deletions(-) diff --git a/backend/app/global_ask_queue.py b/backend/app/global_ask_queue.py index e25f4e0b1..7a154c285 100644 --- a/backend/app/global_ask_queue.py +++ b/backend/app/global_ask_queue.py @@ -134,7 +134,11 @@ async def _run_with_ask_claim_heartbeat( lease: list[object], operation: Any, ) -> Any: - """Renew the claim generation while ``operation`` runs; abort on reclaim.""" + """Renew the claim generation while ``operation`` runs; abort on reclaim. + + If the heartbeat task ends while compute is still running, treat the + owner as lost instead of continuing without renewals. + """ lost = asyncio.Event() stop = asyncio.Event() @@ -154,7 +158,7 @@ async def _beat() -> None: worker = asyncio.create_task(operation) try: await asyncio.wait({worker, beater}, return_when=asyncio.FIRST_COMPLETED) - if lost.is_set(): + if lost.is_set() or (beater.done() and not worker.done()): raise _LostAskClaim() return await worker finally: diff --git a/docs/adr/0370-ask-claim-generation-liveness.md b/docs/adr/0370-ask-claim-generation-liveness.md index f2c33c8d8..be9129ec1 100644 --- a/docs/adr/0370-ask-claim-generation-liveness.md +++ b/docs/adr/0370-ask-claim-generation-liveness.md @@ -29,6 +29,8 @@ the original worker could still settle by job id alone. bounded by the removed 600 s worker deadline. - Live compute is not cancelled when 600 s elapse. Age-based orphan recovery uses three missed heartbeats, not the old 660 s reaper. +- If the heartbeat task ends while compute is still running, abort as a + lost claim rather than continuing without renewals. - Provider `TimeoutError` stays an unavailable Ask failure. ## Consequences diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 8ff70d131..98f8e1c63 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -1,13 +1,13 @@ # Product & Technical Gap Baseline > Ask ownership-fence overlay: 2026-09-08 KST. Issue #975 / #979 exact -> head `29d226529` cleared the settings-collection ImportError (Tests -> run `34152472692`). Follow-up Tests run `34153018910` then failed -> the public-docstring gate on a nested Ask heartbeat helper named -> without a leading underscore. That helper is now private. Frontend -> on that run was GREEN. Explicit 900 s transport timeouts remain -> valid. ADR 0370 Proposed. Not protected-main or independently -> approved. +> head `262d496a9` Tests run `34156704752` was GREEN. Follow-up: a dead +> heartbeat while compute still ran used to leave the owner live without +> renewals (RED `TimeoutError` in +> `test_dead_heartbeat_aborts_live_ask_operation`). The wrapper now +> aborts as a lost claim. Parent #974 Tests GREEN, BLOCKED on +> independent APPROVE. #979 has no independent APPROVE. Do not merge. +> ADR 0370 Proposed. > > Exact-head loop overlay: 2026-08-29 13:20 KST. Protected `main` is > `fc13acaa20adca11968238e398d4aafcf62b6cee` (v2.23.0 leftover-map diff --git a/tests/test_global_ask_claim_cancellation.py b/tests/test_global_ask_claim_cancellation.py index a591e70b7..62faa8928 100644 --- a/tests/test_global_ask_claim_cancellation.py +++ b/tests/test_global_ask_claim_cancellation.py @@ -42,3 +42,42 @@ async def exercise() -> None: ) asyncio.run(exercise()) + + +def test_dead_heartbeat_aborts_live_ask_operation(monkeypatch) -> None: + """A heartbeat that ends while compute is running must not keep the owner live.""" + operation_started = asyncio.Event() + operation_cancelled = asyncio.Event() + + async def operation() -> str: + operation_started.set() + try: + await asyncio.Event().wait() + return "should-not-settle" + finally: + operation_cancelled.set() + + async def exploding_renew(*_args: object, **_kwargs: object) -> None: + raise RuntimeError("heartbeat storage unavailable") + + async def exercise() -> None: + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.01) + monkeypatch.setattr(global_ask_queue, "_renew_ask_claim", exploding_renew) + owner = asyncio.create_task( + global_ask_queue._run_with_ask_claim_heartbeat( + object(), + "job-heartbeat-dead", + [object()], + operation(), + ) + ) + await operation_started.wait() + with pytest.raises(global_ask_queue._LostAskClaim): + await asyncio.wait_for(owner, timeout=1.0) + await asyncio.sleep(0) + assert operation_cancelled.is_set(), ( + "a dead claim heartbeat left compute_global_ask_answer running " + "without renewals" + ) + + asyncio.run(exercise()) From fef48b14f302dd40e1fa83096810cc880a5c4c66 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 8 Sep 2026 07:45:52 +0900 Subject: [PATCH 20/23] docs(ask): reallocate claim-liveness ADR to 0371 Leftover-map validation #980 already allocated ADR 0370 for comparison axis-singular. Keep that number on the leftover stack and record Ask claim-generation liveness as Proposed ADR 0371. --- ...n-liveness.md => 0371-ask-claim-generation-liveness.md} | 7 ++++--- docs/product-technical-gap-baseline.md | 2 +- 2 files changed, 5 insertions(+), 4 deletions(-) rename docs/adr/{0370-ask-claim-generation-liveness.md => 0371-ask-claim-generation-liveness.md} (88%) diff --git a/docs/adr/0370-ask-claim-generation-liveness.md b/docs/adr/0371-ask-claim-generation-liveness.md similarity index 88% rename from docs/adr/0370-ask-claim-generation-liveness.md rename to docs/adr/0371-ask-claim-generation-liveness.md index be9129ec1..7fd51a42f 100644 --- a/docs/adr/0370-ask-claim-generation-liveness.md +++ b/docs/adr/0371-ask-claim-generation-liveness.md @@ -1,11 +1,12 @@ -# ADR 0370 — Global Ask claim-generation liveness +# ADR 0371 — Global Ask claim-generation liveness **Decision status:** Proposed **Date:** 2026-09-08 Amends the Global Ask worker settlement path. Independent of leftover-map -ADRs 0272+ on other stacks and of the versioned translation ledger -([ADR 0362](0362-versioned-ui-translation-ledger.md)). +ADRs 0272–0370 on other stacks (ADR 0370 is leftover-map comparison +axis-singular on the #830 validation lane) and of the versioned +translation ledger ([ADR 0362](0362-versioned-ui-translation-ledger.md)). ## Context diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 98f8e1c63..6727d0b85 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -7,7 +7,7 @@ > `test_dead_heartbeat_aborts_live_ask_operation`). The wrapper now > aborts as a lost claim. Parent #974 Tests GREEN, BLOCKED on > independent APPROVE. #979 has no independent APPROVE. Do not merge. -> ADR 0370 Proposed. +> ADR 0371 Proposed. Leftover-map #980 already holds ADR 0370. > > Exact-head loop overlay: 2026-08-29 13:20 KST. Protected `main` is > `fc13acaa20adca11968238e398d4aafcf62b6cee` (v2.23.0 leftover-map From 06be667c01f930fefb41a5107bf21fceb81ba7e7 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 8 Sep 2026 11:48:42 +0900 Subject: [PATCH 21/23] fix(ask): drain claim renewal before settling completed answers --- backend/app/global_ask_queue.py | 8 +- .../adr/0371-ask-claim-generation-liveness.md | 15 +++ docs/product-requirements.md | 2 +- tests/test_global_ask_claim_cancellation.py | 100 ++++++++++++++++++ tests/test_schema.py | 67 ++++++++++++ 5 files changed, 190 insertions(+), 2 deletions(-) diff --git a/backend/app/global_ask_queue.py b/backend/app/global_ask_queue.py index 7a154c285..e679be88a 100644 --- a/backend/app/global_ask_queue.py +++ b/backend/app/global_ask_queue.py @@ -158,7 +158,13 @@ async def _beat() -> None: worker = asyncio.create_task(operation) try: await asyncio.wait({worker, beater}, return_when=asyncio.FIRST_COMPLETED) - if lost.is_set() or (beater.done() and not worker.done()): + if not worker.done(): + raise _LostAskClaim() + # A renewal may already be committed while its response is in flight. + # Drain it before settlement so the lease contains the committed generation. + stop.set() + (renewal_result,) = await asyncio.gather(beater, return_exceptions=True) + if lost.is_set() or isinstance(renewal_result, BaseException): raise _LostAskClaim() return await worker finally: diff --git a/docs/adr/0371-ask-claim-generation-liveness.md b/docs/adr/0371-ask-claim-generation-liveness.md index 7fd51a42f..33544bd66 100644 --- a/docs/adr/0371-ask-claim-generation-liveness.md +++ b/docs/adr/0371-ask-claim-generation-liveness.md @@ -32,6 +32,12 @@ the original worker could still settle by job id alone. recovery uses three missed heartbeats, not the old 660 s reaper. - If the heartbeat task ends while compute is still running, abort as a lost claim rather than continuing without renewals. +- When compute finishes, stop scheduling renewals and await any renewal + already in flight before admitting either its result or its failure to + settlement. A committed renewal must update the settlement generation; + cancellation during response delivery must not discard that generation. + A failed renewal wins over a simultaneously completed answer. External + owner cancellation still cancels and joins both tasks. - Provider `TimeoutError` stays an unavailable Ask failure. ## Consequences @@ -44,6 +50,15 @@ Negative: crashed workers wait three heartbeat intervals to reclaim. `tests/test_schema.py` proves the settlement compare-and-set against a throwaway database that replayed the real `0001` and `0165` migrations. In-memory queue tests remain for elapsed-time and cancellation contracts. +The completion/renewal race also uses the production queue and renewal +functions against PostgreSQL: delay delivery of an already committed renewal, +finish the answer, then require the stored job to reach `succeeded` with that +answer. This is persistence evidence, not authenticated HTTP/UI acceptance. + +The inherited three-heartbeat recovery ratio is not established as a +deployment capacity or failure-detector contract by these regressions. Its +acceptance remains unresolved; this correction does not calibrate that ratio +or establish that elapsed silence alone proves a dead owner. ## Alternatives considered diff --git a/docs/product-requirements.md b/docs/product-requirements.md index a8b741521..2f0991b61 100644 --- a/docs/product-requirements.md +++ b/docs/product-requirements.md @@ -526,7 +526,7 @@ current boundary until that repository adopts one. | `ContextualWisdomLab/keyverse` | `docs/PRD.md` | Production OIDC/JWKS/identity control plane; local demo Keycloak is not Keyverse | | `ContextualWisdomLab/RankWeave` | No standalone PRD; `README.md`, `ARCHITECTURE.md` | Store-agnostic ranking/fusion dependency; caller owns channels and authorization | | `ContextualWisdomLab/ThreadWeave` | `docs/PRD.md` | Deterministic reference-thread assembly dependency; LineageWeave owns records and persistence | -| `ContextualWisdomLab/DiskSage` | No standalone PRD; `docs/superpowers/specs/2026-07-10-disksage-design.md` | Prospective storage-policy boundary; no current runtime integration | +| `ContextualWisdomLab/disksage` | No standalone PRD; `docs/superpowers/specs/2026-07-10-disksage-design.md` | Prospective storage-policy boundary; no current runtime integration | | `ContextualWisdomLab/wardnet` | No standalone PRD; `README.md`, `docs/architecture.md` | Prospective gateway/network-policy boundary; no current runtime integration | | `ContextualWisdomLab/naruon` | Scoped `docs/topic-intelligence/PRD.md` only | Owns observed calendar/email projections; LineageWeave owns commitments and combined display | | `ContextualWisdomLab/LineageWeave` | This PRD, with ADRs normative | Evidence BI/orchestration, lineage, semantic projection, API, and UI owner | diff --git a/tests/test_global_ask_claim_cancellation.py b/tests/test_global_ask_claim_cancellation.py index 62faa8928..f14a37c64 100644 --- a/tests/test_global_ask_claim_cancellation.py +++ b/tests/test_global_ask_claim_cancellation.py @@ -81,3 +81,103 @@ async def exercise() -> None: ) asyncio.run(exercise()) + + +def test_answer_completion_drains_inflight_claim_renewal(monkeypatch) -> None: + """A committed renewal must reach settlement even if the answer finishes first.""" + renewal_started = asyncio.Event() + answer_finished = asyncio.Event() + renewal_cancelled = asyncio.Event() + original_generation = object() + committed_generation = object() + lease = [original_generation] + + async def renew(*_args): + renewal_started.set() + try: + await answer_finished.wait() + # Model the DB driver's response arriving after the committed UPDATE. + await asyncio.sleep(0) + await asyncio.sleep(0) + return committed_generation + except asyncio.CancelledError: + renewal_cancelled.set() + raise + + async def operation(): + await renewal_started.wait() + answer_finished.set() + return "completed answer" + + async def exercise(): + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.001) + monkeypatch.setattr(global_ask_queue, "_renew_ask_claim", renew) + result = await global_ask_queue._run_with_ask_claim_heartbeat( + object(), "job-renewal-race", lease, operation() + ) + assert result == "completed answer" + assert not renewal_cancelled.is_set() + assert lease[0] is committed_generation + + asyncio.run(exercise()) + + +def test_simultaneous_answer_and_failed_renewal_rejects_answer(monkeypatch) -> None: + """An answer cannot hide a renewal failure in the same event-loop turn.""" + renewal_started = asyncio.Event() + answer_finished = asyncio.Event() + + async def renew(*_args): + renewal_started.set() + await answer_finished.wait() + raise RuntimeError("renewal result unavailable") + + async def operation(): + await renewal_started.wait() + answer_finished.set() + return "answer without confirmed ownership" + + async def exercise(): + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.001) + monkeypatch.setattr(global_ask_queue, "_renew_ask_claim", renew) + with pytest.raises(global_ask_queue._LostAskClaim): + await global_ask_queue._run_with_ask_claim_heartbeat( + object(), "job-simultaneous-completion", [object()], operation() + ) + + asyncio.run(exercise()) + + +def test_owner_cancellation_interrupts_completion_drain(monkeypatch) -> None: + """Draining a renewal never detaches it from native owner cancellation.""" + renewal_started = asyncio.Event() + renewal_cancelled = asyncio.Event() + answer_finished = asyncio.Event() + + async def renew(*_args): + renewal_started.set() + try: + await asyncio.Event().wait() + finally: + renewal_cancelled.set() + + async def operation(): + await renewal_started.wait() + answer_finished.set() + return "completed answer" + + async def exercise(): + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.001) + monkeypatch.setattr(global_ask_queue, "_renew_ask_claim", renew) + owner = asyncio.create_task(global_ask_queue._run_with_ask_claim_heartbeat( + object(), "job-cancel-drain", [object()], operation() + )) + await answer_finished.wait() + await asyncio.sleep(0) + await asyncio.sleep(0) + owner.cancel() + with pytest.raises(asyncio.CancelledError): + await owner + assert renewal_cancelled.is_set() + + asyncio.run(exercise()) diff --git a/tests/test_schema.py b/tests/test_schema.py index b14ec94f9..d970363ee 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -1061,3 +1061,70 @@ def test_cataloged_team_null_affiliation_is_unique(schema_db) -> None: count = cursor.fetchone()[0] assert ids[0] == ids[1] assert count == 1 + + +def test_postgres_ask_completion_retains_inflight_renewal(schema_db, monkeypatch) -> None: + """A real committed renewal cannot strand a completed Ask in running state.""" + from backend.app import global_ask_queue + + with schema_db.cursor() as cur: + cur.execute( + "insert into user_account (external_subject_id, display_name, email_address) " + "values ('ask-renewal-race', 'Synthetic renewal', 'renewal@example.test') " + "returning user_account_id" + ) + account_id = cur.fetchone()[0] + cur.execute( + "insert into global_ask_job " + "(requesting_account_id, question_text, job_status_code) " + "values (%s, 'Synthetic renewal race', 'queued') returning global_ask_job_id", + (account_id,), + ) + job_id = cur.fetchone()[0] + schema_db.commit() + db_dsn = urlunsplit(urlsplit(_ADMIN_DSN)._replace( + path=f"/{schema_db.get_dsn_parameters()['dbname']}" + )) + + async def exercise(): + renewal_committed = asyncio.Event() + answer_finished = asyncio.Event() + renew = global_ask_queue._renew_ask_claim + + async def delayed_renew(*args): + generation = await renew(*args) + assert generation is not None + renewal_committed.set() + await answer_finished.wait() + await asyncio.sleep(0) + await asyncio.sleep(0) + return generation + + async def answer(*_args, **_kwargs): + await renewal_committed.wait() + answer_finished.set() + return {"answer_text": "Synthetic completed answer"} + + async def visibility(*_args): + return set(), set(), False, True + + class _Client: + available = True + + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.001) + monkeypatch.setattr(global_ask_queue, "_renew_ask_claim", delayed_renew) + monkeypatch.setattr(global_ask_queue, "compute_global_ask_answer", answer) + monkeypatch.setattr(global_ask_queue, "load_job_visibility", visibility) + async with asyncpg.create_pool(db_dsn, min_size=1, max_size=2) as pool: + await global_ask_queue.process_global_ask_job( + pool, job_id=str(job_id), chat_factory=_Client + ) + async with pool.acquire() as conn: + row = await conn.fetchrow( + "select job_status_code, answer_payload from global_ask_job " + "where global_ask_job_id = $1", job_id + ) + assert row["job_status_code"] == "succeeded" + assert 'Synthetic completed answer' in row["answer_payload"] + + asyncio.run(exercise()) From 36e3aadd0a6bed3388889ed782462b9173a7a6df Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 8 Sep 2026 11:54:52 +0900 Subject: [PATCH 22/23] docs(gaps): separate current Ask evidence and protected queue state --- docs/product-technical-gap-baseline.md | 137 +++++++++++++++++++++++++ 1 file changed, 137 insertions(+) diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 6727d0b85..f22af25ab 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -1,5 +1,142 @@ # Product & Technical Gap Baseline +## Current evidence — 2026-09-08 KST + +This section supersedes the historical snapshots below. It is a bounded audit, +not a claim that every open PR or product acceptance condition is complete. +Protected main: `83eba56149eb802cd63642c507c324c9976ec78e`. +Reviewed Ask candidate: #979 at `fef48b14f302dd40e1fa83096810cc880a5c4c66`, +stacked on #974 at `def15fc691d4442c0d82103c1642147b1528d7be`. +The completion/renewal correction is source commit +`06be667c01f930fefb41a5107bf21fceb81ba7e7`; this document is its evidence-only +follow-up. Hosted checks and reviews must be fetched for the final PR head, +never inherited from either source commit or parent. + +### Authority and implementation are separate + +- Read the current LineageWeave PRD (`docs/product-requirements.md`) before + choosing the correction. PRD-FR-5 and FR-7 require observable durable outcomes + and preserve the upstream ownership boundary. ADR 0371 is **Proposed**; + accepted ADRs 0204/0213 govern short database transactions and provider work. +- Read contextual-orchestrator `docs/product_planning.md` and + `docs/architecture.md`; its remote main was + `414f22973658c4ddc3d4320fcf7acd9b4e8ba991` at this audit. The Fugu/TRINITY/ + Conductor register supports that owner's orchestration boundary. It does + not establish a LineageWeave timeout, heartbeat ratio, workload SLO, or + model ordering. This correction adds no model-selection or numeric policy. +- Python's official [task/cancellation contract](https://docs.python.org/3/library/asyncio-task.html) + supplies the task lifecycle behavior used by the correction. A task's + completed answer and another task's completion/failure are separate facts. +- GitHub repository API returned canonical names + `ContextualWisdomLab/LineageWeave`, `ContextualWisdomLab/RankWeave`, + `ContextualWisdomLab/ThreadWeave`, `ContextualWisdomLab/disksage`, + `ContextualWisdomLab/TEPP`, and `ContextualWisdomLab/contextual-orchestrator`. + The PRD register now uses the remotely confirmed lowercase `disksage`. +- DeepWiki returned repository-not-indexed; Context7 returned quota-exceeded. + Neither response is documentation or architecture evidence. Sequential + Thinking and Memory graph tools are unavailable in this tool catalog. + +### Selected actionable buyer gap — completed Ask can remain running (#975) + +The reviewed candidate could cancel a claim renewal after PostgreSQL committed +it but before the caller received its generation. A finished answer then used +the old generation for settlement and could stay `running`. A simultaneous +renewal failure could also be ignored when the answer had finished. These are +reproduced lifecycle defects, not inferred performance bottlenecks. They are +prioritized because they prevent an already completed answer from reaching the +requester. No numerical product-gap ranking or population inference is claimed. + +The minimal correction stops scheduling renewals, drains a renewal already in +flight, rejects an unconfirmed/lost claim, and then admits answer or failure to +existing compare-and-set settlement. External cancellation still cancels and +joins both tasks. No API/schema/release-number change, new dependency, provider +call, local mathematical implementation, or UI-copy change is introduced. + +Two focused cases failed before the correction: committed renewal delivery was +cancelled, and a simultaneous renewal error did not reject the answer. After +correction, **42 local tests passed** across claim cancellation, elapsed-time +behavior, queue outcomes, transport configuration, and public docstrings. +The added PostgreSQL regression invokes production claim/renewal/settlement +functions after delaying a real committed renewal response. Its live outcome +is recorded separately below; skipped tests never count as persistence proof. +The inherited three-heartbeat orphan threshold is still ungrounded as a +failure-detector/capacity policy and remains an unresolved ADR acceptance gap. + +### Non-identifying runtime observation, not acceptance + +The formal Compose project remains `lineageweave`. Live Docker mappings bind +backend HTTP to host 18420, contextual-orchestrator to 18000, frontend to 15173, +and PostgreSQL to 15432. No Compose rendering or credentials were printed. + +At 11:41 KST the content-safe k6 harness from exact #964 +`1cced397600b15258b36e221a33beb62c4cca4cd` attempted two VUs for ten seconds, +with an explicit 20-second request observation limit. Synthetic OIDC succeeded +in 16.80 seconds; Ask submission timed out at 19.97 seconds. Setup stopped with +zero active VUs and zero workload iterations. The two setup requests had one +failure (50%) and observed throughput 0.0537 requests/second. These are setup +observations, **not authenticated concurrent workload latency/error/throughput**. +Do not infer whether the timed-out submission eventually persisted or succeeded. + +A nearby one-shot Docker CPU observation was PostgreSQL 175.92%, backend 4.06%, +Ask service 50.83%, Valkey 8.26%, and gateway 67.73%; memory percentages were +2.47%, 1.28%, 1.60%, 8.14%, and 1.69%, respectively. Docker CPU is not normalized +to a single host-core percentage. One snapshot does not prove saturation or a +causal bottleneck; PostgreSQL waits, service occupancy, gateway capacity and +Valkey saturation remain unavailable. No tuning is justified by these numbers. +Authenticated all-page/API acceptance and rendered desktop/mobile acceptance +remain unavailable. There is no UI change in this correction and no new +screenshot or deployed behavior claim. + +### Open queue and protected delivery + +The refreshed GitHub inventory contains **139 open PRs: 17 ready, 122 draft**, +plus **22 open issues**; zero PRs have `reviewDecision=APPROVED`. These describe +the queue, not product maturity. #979's reviewed predecessor has terminal +successful repository backend/frontend Tests, but its only commit check runs +are those two jobs; central receipts and independent approval are not inferred. +#974 has failing/cancelled central review/security receipts as well as the +independent approval requirement. Normal squash auto-merge is enabled on #974; +#780's existing squash auto-merge remains enabled. No merge SHA is claimed. + +Effective main rules require one independent approval, stale-review dismissal, +resolved review threads, and centrally supplied review/security workflows; +force pushes and deletion are prohibited. Classic branch protection returns +404, which does not negate the effective rulesets. No current main or open-PR +run was cancelled. The observed active PR runs belonged to current #983/#966; +no stale closed-PR cancellation was warranted. + +### Collision and Voice acceptance audit + +- The correction retains #979's Proposed ADR 0371. Exact #980 separately adds + ADR 0370 for leftover-map comparison axis-singular evidence; neither number + is allocated anew. #974 changes ADR 0083; #929 owns translation ADR 0362 and + migrations 0246/0247. This correction does not add a migration or release. +- #974 must merge normally before #979 is retargeted to main. Do not enable a + child merge onto its unprotected feature base or inherit parent checks. +- The wider draft report queue still includes duplicate release labels and + serialized overlapping files. A whole-queue ADR/API/schema/release collision + clearance is **not established** by this bounded audit. +- ADR 0246 retains twelve atomic Voices. In the current main, ADR 0251 is the + FJA I/O-psychology document; current #780 names the combination decision + `0256-evidence-bearing-voice-combinations.md` and primary history ADR 0252. + Track title and exact branch as well as number instead of treating the + user's historical ADR 0251 combination reference as the unrelated FJA policy. +- #780 at `1d8fa267b059289e77301a09985dfac70a439814` remains the Voice export + parent. Carrying Post and derivation evidence stay distinct; combinations + remain extensible atomic assignments, with authorized evidence, PROV-O, + truth status and cutoff. Hidden evidence is never substituted. JSON-LD page + subject-property and multi-Voice union candidates #934/#968 remain separate + unmerged evidence, with #937/#971 also pending authorization work. + No fresh authenticated PostgreSQL/API or rendered Voice acceptance was + established here. UI/CSV and paged JSON-LD acceptance remain open. + +## Historical snapshots — not current authority or acceptance + +The preceding file content is retained verbatim below for provenance. Its +head hashes, counts, GREEN claims and timestamps are historical only. + +# Product & Technical Gap Baseline + > Ask ownership-fence overlay: 2026-09-08 KST. Issue #975 / #979 exact > head `262d496a9` Tests run `34156704752` was GREEN. Follow-up: a dead > heartbeat while compute still ran used to leave the owner live without From 6662ea5df6f8f026b54a083677d3e0ee98d2d6d2 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 8 Sep 2026 12:07:31 +0900 Subject: [PATCH 23/23] test(ask): replay required claim-path migrations in schema fixture --- docs/product-technical-gap-baseline.md | 19 ++++++++++++++++--- tests/test_schema.py | 10 ++++++++++ 2 files changed, 26 insertions(+), 3 deletions(-) diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index f22af25ab..d528628d6 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -57,8 +57,17 @@ cancelled, and a simultaneous renewal error did not reject the answer. After correction, **42 local tests passed** across claim cancellation, elapsed-time behavior, queue outcomes, transport configuration, and public docstrings. The added PostgreSQL regression invokes production claim/renewal/settlement -functions after delaying a real committed renewal response. Its live outcome -is recorded separately below; skipped tests never count as persistence proof. +functions after delaying a real committed renewal response. Initial live execution +found that the shared test fixture omitted migrations 0212 and 0218; it failed +with a missing public-verification column before entering the race. The fixture +now applies and replays both real migrations. The normal two-second connection +admission skipped two local tests; skipped tests are not persistence proof. A +separate authenticated PostgreSQL execution with the exact production migrations +passed both regressions: stale-owner settlement rejection and persisted answer +after committed-renewal delivery. Idempotent migration replay also passed. The +throwaway database was dropped afterward. Only visibility/model computation +was synthetic; claim, renewal and settlement used the production functions and +real PostgreSQL. This is database evidence, not authenticated HTTP/UI acceptance. The inherited three-heartbeat orphan threshold is still ungrounded as a failure-detector/capacity policy and remains an unresolved ADR acceptance gap. @@ -128,7 +137,11 @@ no stale closed-PR cancellation was warranted. subject-property and multi-Voice union candidates #934/#968 remain separate unmerged evidence, with #937/#971 also pending authorization work. No fresh authenticated PostgreSQL/API or rendered Voice acceptance was - established here. UI/CSV and paged JSON-LD acceptance remain open. + established here. An exact-head #968 Vitest attempt with one fork could not + start its test process before the runner startup timeout (zero tests ran). + The source diff preserves singleton/array property union, but that inspection + does not replace executable validation. UI/CSV and paged JSON-LD acceptance + remain open. ## Historical snapshots — not current authority or acceptance diff --git a/tests/test_schema.py b/tests/test_schema.py index d970363ee..c95d91464 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -172,6 +172,12 @@ / "migrations" / "0203_global_ask_authorization_scope.sql" ) +_GLOBAL_ASK_CUTOFF_MIGRATION = ( + Path(__file__).resolve().parents[1] / "migrations" / "0212_global_ask_knowledge_cutoff.sql" +) +_GLOBAL_ASK_VERIFICATION_MIGRATION = ( + Path(__file__).resolve().parents[1] / "migrations" / "0218_global_ask_public_verification.sql" +) def _postgres_available() -> bool: @@ -240,10 +246,14 @@ def schema_db(): cur.execute(_SOURCE_EVENT_TIME_MIGRATION.read_text()) cur.execute(_GLOBAL_ASK_JOB_MIGRATION.read_text()) cur.execute(_GLOBAL_ASK_SCOPE_MIGRATION.read_text()) + cur.execute(_GLOBAL_ASK_CUTOFF_MIGRATION.read_text()) + cur.execute(_GLOBAL_ASK_VERIFICATION_MIGRATION.read_text()) # Exercise the production replay contract against the same # PostgreSQL objects instead of merely inspecting SQL text. cur.execute(_GLOBAL_ASK_JOB_MIGRATION.read_text()) cur.execute(_GLOBAL_ASK_SCOPE_MIGRATION.read_text()) + cur.execute(_GLOBAL_ASK_CUTOFF_MIGRATION.read_text()) + cur.execute(_GLOBAL_ASK_VERIFICATION_MIGRATION.read_text()) # Match ADR 0166's production migration executor instead of # maintaining a fixture-owned SQL parser. subprocess.run(