Fix corpus reds: relink provenance, unstamped tier fixtures, slot reaping, wire-receipt cost

devtools/pytest_slot.py

@@ -25,10 +25,11 @@
 import json
 import os
 import shutil
+import signal
 import subprocess
 import sys
 import time
-from collections.abc import Mapping, Sequence
+from collections.abc import Iterator, Mapping, Sequence
 from dataclasses import dataclass
 from pathlib import Path
 from typing import IO, Any, Final
@@ -38,6 +39,7 @@
 __all__ = [
     "BASETEMP_ROOT_ENV",
     "INHERITED_ENVIRONMENT_KEYS",
+    "REAPED_SIGNALS",
     "PYTEST_GROUP",
     "PytestSlotUnavailableError",
     "SlotOutcome",
@@ -59,6 +61,11 @@
 #: The host group whose parallelism is one.
 PYTEST_GROUP: Final = "pytest"
 
+#: Signals that end this process while it waits on a task it owns. The task
+#: outlives the waiter, and the slot's parallelism is one, so an unreaped task
+#: starves every other checkout on the host until someone notices.
+REAPED_SIGNALS: Final[tuple[signal.Signals, ...]] = (signal.SIGINT, signal.SIGTERM, signal.SIGHUP)
+
 #: The only keys the ``pueue add`` client inherits. Everything pytest needs
 #: travels in the launch file, because pueue persists the adder's environment
 #: into shared state.
@@ -212,6 +219,49 @@ def _task_result(status_json: str, task_id: str) -> int:
     raise PytestSlotUnavailableError(REFUSAL.format(reason=f"pueue task {task_id} ended as {result!r}"))
 
 
+def _reap_task(task_id: str, *, env: Mapping[str, str], launch_path: Path | None = None) -> None:
+    """End a task this process owns and drop it from the queue.
+
+    Best effort by construction: the reason we are here is that the waiter is
+    being killed, so a failing reap must not replace the original cause of
+    death with its own error. ``kill`` first because ``remove`` refuses a
+    running task; ``remove`` after because a killed task still occupies the
+    listing. The launch file carries a resolved environment and must not
+    outlive the run either way.
+    """
+
+    for verb in ("kill", "remove"):
+        with contextlib.suppress(PytestSlotUnavailableError):
+            _pueue([verb, task_id], env=env)
+    if launch_path is not None:
+        with contextlib.suppress(OSError):
+            launch_path.unlink(missing_ok=True)
+
+
+@contextlib.contextmanager
+def _reaping(task_id: str, *, env: Mapping[str, str], launch_path: Path | None = None) -> Iterator[None]:
+    """Reap ``task_id`` if this process is signalled or unwound while waiting."""
+
+    def handle(signal_number: int, frame: object) -> None:
+        _reap_task(task_id, env=env, launch_path=launch_path)
+        signal.signal(signal_number, previous.get(signal.Signals(signal_number), signal.SIG_DFL))
+        os.kill(os.getpid(), signal_number)
+
+    previous: dict[signal.Signals, Any] = {}
+    for number in REAPED_SIGNALS:
+        with contextlib.suppress(ValueError, OSError):
+            previous[number] = signal.signal(number, handle)
+    try:
+        yield
+    except BaseException:
+        _reap_task(task_id, env=env, launch_path=launch_path)
+        raise
+    finally:
+        for number, handler in previous.items():
+            with contextlib.suppress(ValueError, OSError):
+                signal.signal(number, handler)
+
+
 def _write_launch(path: Path, *, argv: Sequence[str], cwd: str, env: Mapping[str, str], log_path: Path) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     document = {
@@ -269,9 +319,10 @@ def _queue(
         raise PytestSlotUnavailableError(REFUSAL.format(reason=f"`pueue add` printed no task id: {added.stdout!r}"))
     sys.stderr.write(f"  waiting for the host pytest slot (pueue task {task_id}, group {PYTEST_GROUP}) ...\n")
     sys.stderr.flush()
-    _pueue(["wait", task_id], env=adder)
-    status = _pueue(["status", "--json"], env=adder)
-    returncode = _task_result(status.stdout, task_id)
+    with _reaping(task_id, env=adder, launch_path=launch_path):
+        _pueue(["wait", task_id], env=adder)
+        status = _pueue(["status", "--json"], env=adder)
+        returncode = _task_result(status.stdout, task_id)
     sys.stderr.write(f"  pytest slot released; output: {log_path}\n")
     sys.stderr.flush()
     return SlotOutcome(returncode=returncode, slot=f"pueue task {task_id}", log_path=log_path)

polylogue/storage/attachment_relink.py

@@ -43,12 +43,14 @@
 from polylogue.pipeline.ids import attachment_message_owner_key, message_owner_resolution
 from polylogue.pipeline.services.ingest_worker import IngestRecordResult, SessionWritePayload, ingest_record
 from polylogue.sources.parsers.base import ParsedAttachment, ParsedMessage
+from polylogue.sources.parsers.base_support import derive_attachment_provenance
 from polylogue.storage.runtime.raw.records import RawSessionRecord
 from polylogue.storage.sqlite.archive_tiers.write import (
     _attachment_caption,
     _attachment_id,
     _attachment_message_id_maps,
     _attachment_native_id_values,
+    _attachment_provenance,
     _attachment_reference_positions,
     _attachment_source_url,
     _duplicate_message_native_ids,
@@ -149,6 +151,23 @@ def _message_exists(index_conn: sqlite3.Connection, message_id: str) -> bool:
     return index_conn.execute("SELECT 1 FROM messages WHERE message_id = ?", (message_id,)).fetchone() is not None
 
 
+def _persisted_message_provenance(
+    index_conn: sqlite3.Connection,
+    message_id: str,
+) -> tuple[str | None, str | None]:
+    """Derive an attachment's provenance from the persisted owning turn.
+
+    An append payload's owning message can already be materialized, so it is
+    absent from the payload's own message map. Its role is in the index.
+    """
+
+    row = index_conn.execute("SELECT role, native_id FROM messages WHERE message_id = ?", (message_id,)).fetchone()
+    if row is None:
+        return None, None
+    direction, producer_ref = derive_attachment_provenance(str(row[0]), row[1] or message_id)
+    return direction, producer_ref
+
+
 def _append_materialized_message_ids(
     index_conn: sqlite3.Connection,
     session_id: str,
@@ -341,7 +360,7 @@ def _match_session_payload(
     session_id = payload.session_id
     messages = payload.parsed_session.messages
     position_offset = _next_message_position(index_conn, session_id) if payload.append_only else 0
-    owner_resolution, by_owner_key, _owning_messages = _attachment_message_id_maps(
+    owner_resolution, by_owner_key, owning_messages = _attachment_message_id_maps(
         session_id,
         messages,
         position_offset=position_offset,
@@ -394,14 +413,23 @@ def _match_session_payload(
         message_id = resolved_message_ids.get(id(attachment))
         if message_id is None:
             continue
+        # Provenance goes through the same derivation the writer uses, so a
+        # record parsed before attachments carried direction relinks with the
+        # owning turn's direction rather than a null the column refuses.
+        owning_message = owning_messages.get(message_id)
+        if owning_message is None and attachment.direction is None:
+            direction, producer_ref = _persisted_message_provenance(index_conn, message_id)
+            producer_ref = attachment.producer_ref or producer_ref
+        else:
+            direction, producer_ref = _attachment_provenance(attachment, owning_message, resolved_message_id=message_id)
         recovered[attachment_id] = RelinkableAttachment(
             attachment_id=attachment_id,
             session_id=session_id,
             message_id=message_id,
             position=attachment_positions[id(attachment)],
             upload_origin=attachment.upload_origin,
-            direction=attachment.direction,
-            producer_ref=attachment.producer_ref,
+            direction=direction,
+            producer_ref=producer_ref,
             source_url=_attachment_source_url(attachment),
             caption=_attachment_caption(attachment),
             raw_id=raw_id,

polylogue/storage/sqlite/archive_tiers/write.py

@@ -114,14 +114,25 @@ class ProviderCost:
 
 
 def _write_provider_cost(conn: sqlite3.Connection, session_id: str, model_name: str, cost: ProviderCost) -> None:
-    """Pass through a provider dollar value without catalog computation."""
+    """Pass through a provider dollar value without catalog computation.
+
+    The value is one exact dollar total for the whole session, so it lives on
+    exactly one model row. A merge-append that switches models carries the
+    total to the new row; leaving it on the superseded one would double-count
+    it in every sum over the session.
+    """
     if not isinstance(cost, ProviderCost):
         raise TypeError("provider cost writes require ProviderCost")
     conn.execute(
         """UPDATE session_model_usage SET provider_cost_usd = ?
            WHERE session_id = ? AND model_name = ?""",
         (cost.value, session_id, model_name),
     )
+    conn.execute(
+        """UPDATE session_model_usage SET provider_cost_usd = NULL
+           WHERE session_id = ? AND model_name <> ?""",
+        (session_id, model_name),
+    )
 
 
 logger = get_logger(__name__)

tests/unit/cli/test_embed_activation.py

@@ -36,6 +36,8 @@
     message_window_for_cost,
     read_pending_message_count,
 )
+from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database
+from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
 
 # ---------------------------------------------------------------------------
 # Splicer
@@ -413,7 +415,7 @@ def test_backfill_runs_against_stub_provider(
         )
 
         index_db = tmp_path / "index.db"
-        sqlite3.connect(index_db).close()
+        initialize_archive_database(index_db, ArchiveTier.INDEX)
         pending = [
             PendingSession(session_id="conv-1", title="A", message_count=2),
             PendingSession(session_id="conv-2", title="B", message_count=2),
@@ -457,7 +459,7 @@ def test_backfill_passes_bounded_window_options(
         fake_select = MagicMock(return_value=[])
         fake_preflight = MagicMock(return_value=report)
         index_db = tmp_path / "index.db"
-        sqlite3.connect(index_db).close()
+        initialize_archive_database(index_db, ArchiveTier.INDEX)
         with (
             patch("polylogue.cli.commands.embed._build_preflight_report", fake_preflight),
             patch(
@@ -497,7 +499,7 @@ def test_backfill_routes_archive_to_materializer(
         )
 
         index_db = tmp_path / "index.db"
-        sqlite3.connect(index_db).close()
+        initialize_archive_database(index_db, ArchiveTier.INDEX)
         pending = [PendingSession(session_id="codex-session:v1", title="v1", message_count=2)]
         fake_provider = MagicMock()
         fake_embed = MagicMock(
@@ -545,7 +547,7 @@ def test_backfill_json_outputs_structured_result(
         )
 
         index_db = tmp_path / "index.db"
-        sqlite3.connect(index_db).close()
+        initialize_archive_database(index_db, ArchiveTier.INDEX)
         pending = [PendingSession(session_id="codex-session:v1", title="v1", message_count=2)]
         with (
             _patch_preflight(_make_report(pending_sessions=1, pending_messages=2, max_messages=2)),
@@ -631,7 +633,7 @@ def test_backfill_stop_after_seconds_stops_before_next_session(
         )
 
         index_db = tmp_path / "index.db"
-        sqlite3.connect(index_db).close()
+        initialize_archive_database(index_db, ArchiveTier.INDEX)
         # Event-driven fake clock: scripted read sequences broke twice (a
         # StopIteration when the command read the clock an extra time, then a
         # second embed when an extra read consumed the past-deadline value).
@@ -692,7 +694,7 @@ def test_backfill_max_errors_stops_after_provider_error(
         )
 
         index_db = tmp_path / "index.db"
-        sqlite3.connect(index_db).close()
+        initialize_archive_database(index_db, ArchiveTier.INDEX)
         fake_embed = MagicMock(
             side_effect=[
                 EmbedSessionOutcome(status="error", session_id="conv-1", error="provider 429"),
@@ -740,7 +742,7 @@ def test_backfill_run_cost_cap_stops_before_provider_call(
         )
 
         index_db = tmp_path / "index.db"
-        sqlite3.connect(index_db).close()
+        initialize_archive_database(index_db, ArchiveTier.INDEX)
         fake_embed = MagicMock(
             side_effect=[
                 EmbedSessionOutcome(status="embedded", session_id="conv-1", embedded_message_count=2),

tests/unit/core/test_insight_readiness.py

@@ -206,6 +206,7 @@ async def test_insight_readiness_report_marks_missing_insight_tables(tmp_path: P
                 tool_name TEXT,
                 tool_result_exit_code INTEGER,
                 tool_result_is_error INTEGER,
+                tool_outcome TEXT,
                 search_text TEXT
             );
             INSERT INTO sessions (session_id, parent_session_id, source_name, sort_key, updated_at)

tests/unit/core/test_synthetic_wire_support.py

@@ -167,7 +167,7 @@ def drop_first_coverage_witness(
         )
 
     monkeypatch.setattr(dispatch_module, "parse_payload", drop_first_coverage_witness)
-    receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry())
+    receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry(), providers=("chatgpt",))
 
     entry = next(item for item in receipt.entries if item.provider == "chatgpt")
     dropped = next(witness for witness in entry.parser_witnesses if witness.index == 0)
@@ -751,7 +751,7 @@ def fail_baseline(
         )
 
     monkeypatch.setattr(dispatch_module, "parse_payload", fail_baseline)
-    receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry())
+    receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry(), providers=("chatgpt",))
 
     entry = next(item for item in receipt.entries if item.provider == "chatgpt")
     baseline = next(item for item in entry.parser_witnesses if item.artifact_kind == "baseline")
@@ -775,7 +775,7 @@ def fail_chatgpt_baseline(self: SchemaValidator, payload: JSONValue) -> validato
         return original_validate(self, payload)
 
     monkeypatch.setattr(validator_module.SchemaValidator, "validate", fail_chatgpt_baseline)
-    receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry())
+    receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry(), providers=("chatgpt",))
 
     entry = next(item for item in receipt.entries if item.provider == "chatgpt")
     baseline = next(item for item in entry.parser_witnesses if item.artifact_kind == "baseline")
@@ -892,11 +892,19 @@ def test_antigravity_metadata_only_source_has_no_language_server_session(tmp_pat
     assert sessions == []
 
 
+#: Building the receipt runs the full synthetic corpus and parser for every
+#: catalogued package element, so a test that asserts a receipt *changed*
+#: names the providers whose entries carry the change rather than paying for
+#: the whole catalog. Catalog-wide construction stays covered by
+#: ``test_support_receipt_is_deterministic``.
+_COVERAGE_PROVIDERS = ("chatgpt", "codex")
+
+
 def test_construct_handler_removal_changes_coverage_receipt(monkeypatch: pytest.MonkeyPatch) -> None:
-    before = wire_formats.build_wire_support_receipt(registry=SchemaRegistry())
+    before = wire_formats.build_wire_support_receipt(registry=SchemaRegistry(), providers=_COVERAGE_PROVIDERS)
 
     monkeypatch.delitem(SCHEMA_CONSTRUCT_HANDLERS, "array")
-    after = wire_formats.build_wire_support_receipt(registry=SchemaRegistry())
+    after = wire_formats.build_wire_support_receipt(registry=SchemaRegistry(), providers=_COVERAGE_PROVIDERS)
 
     assert before.to_dict() != after.to_dict()
     assert not after.complete
@@ -1032,7 +1040,7 @@ def __init__(
             super().__init__(schema, strict=strict, provider=provider)
 
     monkeypatch.setattr(validator_module, "SchemaValidator", CapturingValidator)
-    receipt = wire_formats.build_wire_support_receipt(registry=registry)
+    receipt = wire_formats.build_wire_support_receipt(registry=registry, providers=("chatgpt",))
 
     entry = next(item for item in receipt.entries if item.provider == "chatgpt")
     assert entry.schema_valid is True
@@ -1045,7 +1053,7 @@ def __init__(
 
 
 def test_claude_code_route_only_waives_unrepresentable_nested_content() -> None:
-    receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry())
+    receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry(), providers=("claude-code",))
     entry = next(item for item in receipt.entries if item.provider == "claude-code")
 
     assert entry.construct_coverage is not None
@@ -1147,10 +1155,10 @@ def test_missing_required_and_additional_property_evidence_makes_receipt_incompl
 
 
 def test_removed_provider_route_changes_explicit_support_receipt(monkeypatch: pytest.MonkeyPatch) -> None:
-    before = wire_formats.build_wire_support_receipt(registry=SchemaRegistry())
+    before = wire_formats.build_wire_support_receipt(registry=SchemaRegistry(), providers=_COVERAGE_PROVIDERS)
 
     monkeypatch.delitem(wire_formats.PROVIDER_WIRE_ROUTES, "codex")
-    after = wire_formats.build_wire_support_receipt(registry=SchemaRegistry())
+    after = wire_formats.build_wire_support_receipt(registry=SchemaRegistry(), providers=_COVERAGE_PROVIDERS)
 
     assert before.to_dict() != after.to_dict()
     assert after.missing_routes == ("codex",)