Skip to content

test(storage): prove raw-authority fair scheduling and CAS-typed retry (hjpx AC3/AC4) #3345

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Sinity merged 1 commit into from
Jul 27, 2026
Merged

test(storage): prove raw-authority fair scheduling and CAS-typed retry (hjpx AC3/AC4) #3345

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
197 changes: 197 additions & 0 deletions tests/unit/storage/test_repair.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2249,6 +2249,79 @@ def acquisition_only_order(candidates: Any, *, archive_root: Path) -> list[tuple
assert unfair_second == unfair_first


def test_raw_materialization_ordering_is_size_agnostic_and_does_not_starve_large_work(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""hjpx AC3: bounded scheduling must pick components by stable
fairness/age, not by cheapness -- a size-preferring order recreates the
exact starvation failure mode AC3 names ("repeatedly selecting the same
cheap components... starving large valid work"), even though every
component here is independently executable in one pass.
"""
from polylogue.core.enums import Provider
from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore
from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root

def run(*, prefer_cheap: bool) -> tuple[tuple[str, ...], str]:
root = tmp_path / ("cheap-first" if prefer_cheap else "fair-order")
initialize_active_archive_root(root)
with ArchiveStore.open_existing(root, read_only=False) as archive:
# The large valid component is acquired FIRST (oldest), so fair
# age-based ordering must select it on the very first pass.
large_raw_id = archive.write_raw_payload(
provider=Provider.CODEX,
payload=b'{"type":"session_meta","payload":{"id":"large-valid"}}\n',
source_path="large-valid.jsonl",
acquired_at_ms=1,
)
for index in range(5):
archive.write_raw_payload(
provider=Provider.CODEX,
payload=f'{{"type":"session_meta","payload":{{"id":"cheap-{index}"}}}}\n'.encode(),
source_path=f"cheap-{index}.jsonl",
acquired_at_ms=index + 2,
)
with sqlite3.connect(root / "source.db") as source_conn:
# blob_size is scheduling metadata only (parsing reads the tiny
# real payload); this makes the large component "expensive but
# still executable" (well under the execute limit) without
# generating megabytes of fixture bytes.
source_conn.execute(
"UPDATE raw_sessions SET blob_size = ? WHERE raw_id = ?",
(repair_mod.RAW_MATERIALIZATION_EXECUTE_BLOB_LIMIT_BYTES // 2, large_raw_id),
)
source_conn.commit()

config = _config(root)
_complete_bounded_raw_census(config, limit=1)
with monkeypatch.context() as mutation:
if prefer_cheap:

def cheap_first_order(candidates: Any, *, archive_root: Path) -> list[tuple[str, ...]]:
candidate_ids = set(candidates.raw_ids)
source_components = candidates.authority_components or tuple(
(raw_id,) for raw_id in candidates.raw_ids
)
components = [c for c in source_components if candidate_ids.intersection(c)]
return sorted(
components,
key=lambda component: sum(
candidates.expanded_blob_bytes.get(rid, candidates.raw_blob_bytes.get(rid, 0))
for rid in component
),
)

mutation.setattr(repair_mod, "_raw_materialization_ordered_components", cheap_first_order)
result = repair_mod.repair_raw_materialization(config, raw_artifact_limit=1)
return result.plan_outcomes[0].input_raw_ids, large_raw_id

fair_selected, fair_large_id = run(prefer_cheap=False)
cheap_selected, cheap_large_id = run(prefer_cheap=True)

assert fair_selected == (fair_large_id,)
assert cheap_selected != (cheap_large_id,)


def test_raw_materialization_isolates_failed_component_and_continues_batch(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
Expand Down Expand Up @@ -2285,6 +2358,130 @@ def fail_oldest(*args: Any, selected_raw_ids: list[str] | None = None, **kwargs:
assert [outcome.status.value for outcome in result.plan_outcomes].count("executed") == 2


def test_raw_materialization_transient_failure_retries_with_same_plan_id_then_succeeds(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""hjpx AC4: a transient interruption must remain retryable under the
*same* plan id and later succeed once -- not spawn a fresh plan id, and
not silently mutate anything before the retry lands.
"""
from polylogue.core.enums import Provider
from polylogue.sources import revision_backfill
from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore
from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root

initialize_active_archive_root(tmp_path)
with ArchiveStore.open_existing(tmp_path, read_only=False) as archive:
raw_id = archive.write_raw_payload(
provider=Provider.CODEX,
payload=b'{"type":"session_meta","payload":{"id":"transient-target"}}\n',
source_path="transient-target.jsonl",
acquired_at_ms=1,
)

original = revision_backfill.backfill_historical_revision_evidence
should_fail = True

def fail_once(*args: Any, selected_raw_ids: list[str] | None = None, **kwargs: Any) -> Any:
if should_fail and selected_raw_ids == [raw_id]:
raise RuntimeError("OperationalError: database is locked")
return original(*args, selected_raw_ids=selected_raw_ids, **kwargs)

monkeypatch.setattr(revision_backfill, "backfill_historical_revision_evidence", fail_once)

config = _config(tmp_path)
first = repair_mod.repair_raw_materialization(config)
assert first.plan_outcomes[0].status.value == "retryable"
assert "database is locked" in first.plan_outcomes[0].reason
first_plan_id = first.plan_outcomes[0].plan_id

# Non-mutating: the injected failure must not have left any parse/apply
# residue behind before the retry runs.
with sqlite3.connect(tmp_path / "source.db") as source_conn:
assert source_conn.execute("SELECT parsed_at_ms FROM raw_sessions WHERE raw_id = ?", (raw_id,)).fetchone() == (
None,
)

should_fail = False
second = repair_mod.repair_raw_materialization(config)

assert second.plan_outcomes[0].status.value == "executed"
assert second.plan_outcomes[0].plan_id == first_plan_id
with sqlite3.connect(tmp_path / "source.db") as source_conn:
assert source_conn.execute(
"SELECT parsed_at_ms IS NOT NULL FROM raw_sessions WHERE raw_id = ?", (raw_id,)
).fetchone() == (1,)


def test_raw_materialization_cas_conflict_outcome_is_typed_durable_and_non_mutating(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""hjpx AC4: a CAS conflict/incomparable-authority rejection from the
revision-application layer must surface as a typed, durably-recorded,
non-mutating outcome through the reconciler -- it must not silently
vanish, apply partially, or lose its plan id.
"""
from polylogue.core.enums import Provider
from polylogue.sources import revision_backfill
from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore
from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root

initialize_active_archive_root(tmp_path)
with ArchiveStore.open_existing(tmp_path, read_only=False) as archive:
raw_id = archive.write_raw_payload(
provider=Provider.CODEX,
payload=b'{"type":"session_meta","payload":{"id":"cas-conflict-target"}}\n',
source_path="cas-conflict-target.jsonl",
acquired_at_ms=1,
)

cas_message = (
"raw revision CAS rejected a conflicting accepted head: "
"logical_source_key='codex:cas-conflict-target' existing(session_id='cas-conflict-target', "
"accepted_raw_id='other-raw') incoming(session_id='cas-conflict-target', accepted_raw_id='" + raw_id + "')"
)

def raise_cas_conflict(*args: Any, selected_raw_ids: list[str] | None = None, **kwargs: Any) -> Any:
assert selected_raw_ids == [raw_id]
raise RuntimeError(cas_message)
Comment on lines +2444 to +2446

@coderabbitai coderabbitai Bot Jul 27, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Confirm the per-component backfill_historical_revision_evidence call in repair_raw_materialization
# is wrapped in a broad except that stringifies exceptions into the outcome reason.
rg -n 'backfill_historical_revision_evidence\(' -B5 -A25 polylogue/storage/repair.py

Repository: Sinity/polylogue

Length of output: 1936

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Map the test and implementation around the relevant scope without running repository code.
printf '--- test_repair outline matches ---\n'
fd -a 'test_repair.py' . | while read -r f; do
  wc -l "$f"
  ast-grep outline "$f" --match 'test_raw_materialization_transient_failure_retries_with_same_plan_id_then_succeeds' --view compact 2>/dev/null || true
done

printf '\n--- tests/unit/storage/test_repair.py lines 2360-2465 ---\n'
sed -n '2360,2465p' tests/unit/storage/test_repair.py | nl -ba -v2360

printf '\n--- polylogue/storage/repair.py lines 6460-6545 ---\n'
sed -n '6460,6545p' polylogue/storage/repair.py | nl -ba -v6460

printf '\n--- deterministic except shape probe from source ---\n'
python3 - <<'PY'
from pathlib import Path
p = Path('polylogue/storage/repair.py')
s = p.read_text()
start = s.index('for plan, raw_id in zip(executable_plans, executable_raw_ids, strict=True):')
end = s.index('return RawReplayPlanResults', start)
chunk = s[start:end]
print('contains bare try:', 'try:' in chunk)
print('except Exception or base Exception:', 'except Exception' in chunk or 'except BaseException' in chunk)
print('contains AttributeError specifically before second except:')
for i, idx in enumerate(chunk.find('except ') for _ in range(10)):
    pass
# Locate except bodies after the critical try
except_lines = []
for m in __import__('re').finditer(r'except[^\n]*:(?P<body>(?:\n\s+.*)*)', chunk):
    body = m.group('body').splitlines()
    if body and body[0].strip():
        first = body[0].strip()
        if len(first) < 160 and first.startswith('outcome'):
            except_lines.append((m.group().splitlines()[0].strip(), first))
print('outcome assignments in except bodies:')
for line, first in except_lines:
    print(line)
    print(first)
PY

Repository: Sinity/polylogue

Length of output: 245

🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf '%s\n' '--- test_repair outline matches ---'
fd -a 'test_repair.py' . | while read -r f; do
  wc -l "$f"
  ast-grep outline "$f" --match 'test_raw_materialization_transient_failure_retries_with_same_plan_id_then_succeeds' --view compact 2>/dev/null || true
done

printf '%s\n' ''
printf '%s\n' '--- tests/unit/storage/test_repair.py lines 2360-2465 ---'
sed -n '2360,2465p' tests/unit/storage/test_repair.py | nl -ba -v2360

printf '%s\n' ''
printf '%s\n' '--- polylogue/storage/repair.py lines 6460-6545 ---'
sed -n '6460,6545p' polylogue/storage/repair.py | nl -ba -v6460

printf '%s\n' ''
printf '%s\n' '--- deterministic except shape probe from source ---'
python3 - <<'PY'
from pathlib import Path
import re

p = Path('polylogue/storage/repair.py')
s = p.read_text()
start = s.index('for plan, raw_id in zip(executable_plans, executable_raw_ids, strict=True):')
end = s.index('return RawReplayPlanResults', start)
chunk = s[start:end]

print('contains bare try:', 'try:' in chunk)
print('contains except Exception:', 'except Exception' in chunk)
print('contains except BaseException:', 'except BaseException' in chunk)

except_lines = []
for m in re.finditer(r'except[^\n]*:(?P<body>(?:\n\s+.*)*)', chunk):
    body = m.group('body').splitlines()
    if body and body[0].strip().startswith('outcome = RawReplayPlanOutcome'):
        first = body[0].strip()
        except_lines.append((m.group().splitlines()[0].strip(), first))

print('outcome assignments in except bodies:')
for line, first in except_lines:
    print(line)
    print(first)
PY

Repository: Sinity/polylogue

Length of output: 348

🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf '%s\n' '--- tests/unit/storage/test_repair.py lines 2360-2465 ---'
sed -n '2360,2465p' tests/unit/storage/test_repair.py

printf '%s\n' ''
printf '%s\n' '--- polylogue/storage/repair.py lines 6460-6545 ---'
sed -n '6460,6545p' polylogue/storage/repair.py

printf '%s\n' ''
printf '%s\n' '--- deterministic except shape probe from source ---'
python3 - <<'PY'
from pathlib import Path
import re

p = Path('polylogue/storage/repair.py')
s = p.read_text()
start = s.index('for plan, raw_id in zip(executable_plans, executable_raw_ids, strict=True):')
end = s.index('return RawReplayPlanResults', start)
chunk = s[start:end]

print('contains bare try:', 'try:' in chunk)
print('contains except Exception:', 'except Exception' in chunk)
print('contains except BaseException:', 'except BaseException' in chunk)

except_lines = []
for m in re.finditer(r'except[^\n]*:(?P<body>(?:\n\s+.*)*)', chunk):
    body = m.group('body').splitlines()
    if body and body[0].strip().startswith('outcome = RawReplayPlanOutcome'):
        first = body[0].strip()
        except_lines.append((m.group().splitlines()[0].strip(), first))

print('outcome assignments in except bodies:')
for line, first in except_lines:
    print(line)
    print(first)
PY

Repository: Sinity/polylogue

Length of output: 10222

Move the selected_raw_ids invariant outside the patched call

repair_raw_materialization wraps backfill_historical_revision_evidence() in except Exception and converts runtime failures to a RETRYABLE outcome with reason=f"component execution raised {type(exc).__name__}: {exc}". If assert selected_raw_ids == [raw_id] fails, the test surfaces as a failed retryable run instead of a clear assertion failure. Record the call arg and assert after repair_raw_materialization returns.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/unit/storage/test_repair.py` around lines 2444 - 2446, Update the test
stub raise_cas_conflict to record selected_raw_ids instead of asserting inside
the patched call, then assert the recorded value equals [raw_id] after
repair_raw_materialization returns. Preserve the RuntimeError(cas_message)
behavior so the production exception-to-RETRYABLE path remains exercised.


monkeypatch.setattr(revision_backfill, "backfill_historical_revision_evidence", raise_cas_conflict)
result = repair_mod.repair_raw_materialization(_config(tmp_path))

outcome = result.plan_outcomes[0]
assert outcome.status.value == "retryable"
assert "CAS rejected a conflicting accepted head" in outcome.reason

assert result.census_receipt is not None
with sqlite3.connect(tmp_path / "source.db") as source_conn:
# Durable: the typed outcome is recorded against the exact plan id in
# the durable source-tier ledger, not only in the in-memory receipt.
recorded = source_conn.execute(
"""
SELECT outcome_status, reason
FROM raw_authority_census_plans
WHERE census_id = ? AND plan_id = ?
""",
(result.census_receipt.census_id, outcome.plan_id),
).fetchone()
assert recorded == ("retryable", outcome.reason)
# Non-mutating: no parse residue exists for the raw the CAS
# rejection blocked.
assert source_conn.execute("SELECT parsed_at_ms FROM raw_sessions WHERE raw_id = ?", (raw_id,)).fetchone() == (
None,
)
with sqlite3.connect(tmp_path / "index.db") as index_conn:
# Non-mutating: no application/head state exists in the (rebuildable
# but still write-through) index tier either.
assert (
index_conn.execute("SELECT COUNT(*) FROM raw_revision_applications WHERE raw_id = ?", (raw_id,)).fetchone()[
0
]
== 0
)
assert index_conn.execute("SELECT COUNT(*) FROM raw_revision_heads").fetchone()[0] == 0


def test_raw_materialization_fails_closed_on_plan_conservation_mismatch(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
Expand Down