Skip to content

fix(storage): dedupe judgment-request generation across census cycles #3290

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

fix(storage): dedupe judgment-request generation across census cycles #3290

Show file tree
Hide file tree
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
32 changes: 30 additions & 2 deletions polylogue/storage/raw_reconciler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -649,13 +649,41 @@ def _strategy_overrides(


def _record_judgment_candidate(config: Config, item: RawAuthorityFrontierItem, *, now_ms: int) -> tuple[str, bool]:
"""Persist the conflict as a non-authoritative candidate for operator judgment."""
"""Persist the conflict as a non-authoritative candidate for operator judgment.

polylogue-rjtv: the assertion id is derived from ``item.plan_id``, which is
itself derived from a fresh evidence digest every census cycle -- a
census cycle that re-encounters the *same* unresolved conflict (same
``raw_id``/``logical_source_key``) before an operator has judged it would
otherwise mint a brand-new candidate each time, leaving prior cycles'
still-pending duplicates to accumulate forever (found live 2026-07-27: 24
candidates in ``judge --list`` for what was actually 6 real conflicts).
Look up an existing still-``candidate`` request for the same conflict
identity first and refresh it in place instead of minting a new one. This
only dedupes pending-vs-pending; an already accepted/rejected/deferred
assertion is untouched, so a fresh judgment can still be requested if the
same conflict resurfaces after a prior disposition.
"""
from polylogue.core.enums import AssertionKind, AssertionStatus, AssertionVisibility
from polylogue.storage.sqlite.archive_tiers.user_write import read_assertion_envelope, upsert_assertion

assertion_id = f"judgment:{_digest(['raw-authority-frontier', item.plan_id])}"
root = _archive_root(config)
with closing(sqlite3.connect(root / "user.db")) as conn, conn:
pending_row = conn.execute(
"""
SELECT assertion_id FROM assertions
WHERE kind = 'judgment' AND status = 'candidate'
AND json_extract(value_json, '$.raw_id') = ?
AND json_extract(value_json, '$.logical_source_key') = ?
LIMIT 1
""",
(item.raw_id, item.logical_source_key),
Comment on lines +676 to +680

@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.

🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

Handle logical_source_key=None in the identity lookup.

RawAuthorityFrontierItem.logical_source_key is nullable, but json_extract(...) = ? with None evaluates to SQL NULL, never true. Repeated conflicts without a logical-source key will still create duplicates. Use SQLite null-safe comparison (IS ?) and add a nullable-key regression case.

🤖 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 `@polylogue/storage/raw_reconciler.py` around lines 676 - 680, Update the
identity lookup query in the raw reconciliation flow to use SQLite’s null-safe
IS comparison for logical_source_key, allowing None to match stored JSON null
values while preserving non-null matching. Add a regression test covering
repeated conflicts where RawAuthorityFrontierItem.logical_source_key is None and
verify no duplicate is created.

).fetchone()
assertion_id = (
str(pending_row[0])
if pending_row is not None
else f"judgment:{_digest(['raw-authority-frontier', item.plan_id])}"
)
Comment on lines +672 to +686

@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.

🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

rg -n -C 6 \
  'def _immediate_user_write_transaction|CREATE (UNIQUE )?INDEX|CREATE TABLE assertions' \
  polylogue/storage/sqlite

rg -n -C 8 \
  'pending_row = conn\.execute|def _record_judgment_candidate|def upsert_assertion' \
  polylogue/storage/raw_reconciler.py \
  polylogue/storage/sqlite/archive_tiers/user_write.py

Repository: Sinity/polylogue

Length of output: 50373

🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf '== Files matching raw_reconciler.py/user_write.py/assertions DDL ==\n'
fd -a 'raw_reconciler.py|user_write.py' polylogue/storage | sed 's#^\./##'
rg -n -C 4 'CREATE TABLE assertions|CREATE UNIQUE INDEX.*assertions|CREATE INDEX.*assertions' \
  polylogue/storage/sqlite/migrations polylogue/storage/sqlite/archive_tiers \
  --max-count 80

printf '\n== raw_reconciler relevant sections ==\n'
wc -l polylogue/storage/raw_reconciler.py
sed -n '620,710p' polylogue/storage/raw_reconciler.py
printf '\n== upsert_assertion usages and definitions ==\n'
rg -n -C 12 'def upsert_assertion|upsert_assertion|_record_judgment_candidate|pending_row = conn\.execute' \
  polylogue/storage/raw_reconciler.py polylogue/storage/sqlite/archive_tiers/user_write.py \
  --max-count 100

Repository: Sinity/polylogue

Length of output: 50372

🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf '== user_write immediate transaction and upsert_assertion implementation ==\n'
sed -n '86,130p' polylogue/storage/sqlite/archive_tiers/user_write.py
sed -n '1099,1220p' polylogue/storage/sqlite/archive_tiers/user_write.py

printf '\n== assertions table DDL and all assertion indexes/migrations ==\n'
python3 - <<'PY'
from pathlib import Path
import re
root = Path('polylogue/storage/sqlite')
for p in sorted(root.glob('**/*.sql')) + sorted(root.glob('archive_tiers/*.py')):
    text = p.read_text()
    if 'CREATE TABLE assertions' in text or 'idx_assertions' in text:
        print(f'--- {p} ---')
        for line in re.finditer(r'(?ms)CREATE TABLE assertions[\s\S]+?(?=\n\n|\Z)', text):
            end = line.end()
            start=max(0,end-800); print(text[start:end][:1200])
        for l in re.finditer(r'(?m)CREATE .* INDEX .* assertions|idx_assertions', text):
            start=max(0,l.start()-150); end=min(len(text),l.end()+150)
            print(text[start:end])
PY

printf '\n== call sites around _record_judgment_candidate and transaction context ==\n'
rg -n -C 6 '_record_judgment_candidate|with closing|conn:|BEGIN IMMEDIATE|_immediate_user_write_transaction' \
  polylogue/storage/raw_reconciler.py polylogue/storage -g '*.py' \
  --max-count 120

Repository: Sinity/polylogue

Length of output: 50373

🏁 Script executed:

#!/bin/bash
set -euo pipefail

python3 - <<'PY'
from pathlib import Path
import re

for p in [Path('polylogue/storage/sqlite/archive_tiers/user.py'), Path('polylogue/storage/sqlite/archive_tiers/user_write.py')]:
    text = p.read_text()
    for name in ['CREATE TABLE assertions', 'CREATE UNIQUE INDEX', 'idx_assertions']:
        if name in text:
            print(f'--- assertions occurrences in {p} ---')
            lines = text.splitlines()
            for i,l in enumerate(lines,1):
                if name in l:
                    lo=max(1,i-3); hi=min(len(lines),i+20 if 'idx_assertions' in l else i+8)
                    print(f'[{p}:{lo}-{hi}]')
                    for j in range(lo,hi+1):
                        print(f'{j:4}: {lines[j-1]}')
                    print()

print('--- focused assertions index/statements by regex ---')
for p in [Path('polylogue/storage/sqlite/archive_tiers/user.py'), Path('polylogue/storage/sqlite/archive_tiers/user_write.py')]:
    text = p.read_text()
    for m in re.finditer(r'(?ms)CREATE (UNIQUE )?INDEX\s+\Qidx_assertions\E[\s\S]{0,220}', text):
        print('---', p, '---')
        print(m.group(0))
PY

printf '\n===== raw_reconciler _record_judgment_candidate complete =====\n'
sed -n '640,705p' polylogue/storage/raw_reconciler.py

printf '\n===== user_write immediate transaction and upsert_assertion =====\n'
sed -n '86,120p' polylogue/storage/sqlite/archive_tiers/user_write.py
sed -n '1099,1195p' polylogue/storage/sqlite/archive_tiers/user_write.py

printf '\n===== all candidate creation paths that derive assertion IDs outside immediate_user_write_transaction =====\n'
sed -n '1215,1295p' polylogue/storage/sqlite/archive_tiers/user_write.py
sed -n '1315,1630p' polylogue/storage/sqlite/archive_tiers/user_write.py

Repository: Sinity/polylogue

Length of output: 6326

🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf '===== raw_reconciler _record_judgment_candidate complete =====\n'
sed -n '640,705p' polylogue/storage/raw_reconciler.py

printf '\n===== user_write immediate transaction and upsert_assertion =====\n'
sed -n '86,120p' polylogue/storage/sqlite/archive_tiers/user_write.py
sed -n '1099,1195p' polylogue/storage/sqlite/archive_tiers/user_write.py

printf '\n===== assertions table DDL and indexes from user.py =====\n'
sed -n '20,62p' polylogue/storage/sqlite/archive_tiers/user.py

printf '\n===== candidate creation paths that derive assertion IDs outside immediate_user_write_transaction =====\n'
sed -n '1215,1290p' polylogue/storage/sqlite/archive_tiers/user_write.py
sed -n '1315,1630p' polylogue/storage/sqlite/archive_tiers/user_write.py

Repository: Sinity/polylogue

Length of output: 29642

Make raw-authority candidate creation atomic.

_record_judgment_candidate selects an existing candidate row outside the upsert_assertion immediate transaction, so two concurrent census cycles can both miss, derive the same item.plan_id assertion id, and both insert. Wrap the lookup-and-upsert in one write transaction, or add a durable conflict-identity invariant for the pending raw-authority frontier assertion and handle its conflict path.

🤖 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 `@polylogue/storage/raw_reconciler.py` around lines 672 - 686, Make the lookup
and creation in _record_judgment_candidate atomic by moving the pending-row
SELECT and subsequent upsert_assertion call into the same immediate write
transaction. Preserve the existing lookup criteria and deterministic
assertion_id fallback, and ensure concurrent census cycles resolve through the
transaction/conflict path without duplicate candidate rows.

existing = read_assertion_envelope(conn, assertion_id)
if existing is not None and existing.status is not AssertionStatus.CANDIDATE:
return existing.assertion_id, False
Expand Down
45 changes: 45 additions & 0 deletions tests/unit/storage/test_browser_capture_origin_repair.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,9 @@
from polylogue.storage.raw_authority import resolve_raw_authority_blocker
from polylogue.storage.raw_reconciler import (
RawAuthorityActuator,
RawAuthorityFrontierItem,
RawAuthorityFrontierState,
_record_judgment_candidate,
apply_raw_authority_frontier,
inspect_raw_authority_frontier,
)
Expand Down Expand Up @@ -774,6 +776,49 @@ def test_unified_frontier_conflict_requires_typed_judgment_then_resumes_same_evi
).fetchone() == (0,)


def test_repeat_census_of_same_pending_conflict_reuses_one_judgment_candidate(tmp_path: Path) -> None:
"""polylogue-rjtv: two census cycles hitting the same still-unjudged
conflict (same raw_id/logical_source_key, different plan_id/evidence -
exactly what a fresh census run produces each time it re-derives
evidence) must not mint a second pending judgment candidate. Found live
2026-07-27: 24 candidates in ``judge --list`` for what was actually 6
real conflicts, because assertion_id was derived from the ephemeral
plan_id instead of the stable conflict identity."""
initialize_active_archive_root(tmp_path)

def _item(plan_suffix: str) -> RawAuthorityFrontierItem:
return RawAuthorityFrontierItem(
state=RawAuthorityFrontierState.CONFLICTING_AUTHORITY_NEEDS_JUDGMENT,
actuator=RawAuthorityActuator.REQUEST_JUDGMENT,
raw_id="raw-rjtv-shared",
logical_source_key="unknown:rjtv-conflict",
session_id="chatgpt-export:rjtv-conflict",
reason="byte-proven browser rekey requires no retained membership census",
evidence_digest=f"digest-{plan_suffix}",
input_raw_ids=("raw-rjtv-shared",),
source_preconditions={},
index_preconditions={},
strategy_witness={"kind": "browser_conflict"},
plan_id=f"raw-authority-frontier:{plan_suffix}",
)

config = _config(tmp_path)
first_id, _ = _record_judgment_candidate(config, _item("cycle-one"), now_ms=1000)
second_id, _ = _record_judgment_candidate(config, _item("cycle-two"), now_ms=2000)

assert second_id == first_id
with sqlite3.connect(tmp_path / "user.db") as user:
user.row_factory = sqlite3.Row
rows = user.execute(
"SELECT assertion_id, value_json FROM assertions WHERE kind = 'judgment' AND status = 'candidate'"
).fetchall()
assert len(rows) == 1
assert rows[0]["assertion_id"] == first_id
# The single surviving row reflects the LATEST cycle's plan, not stale
# evidence from the first.
assert json.loads(rows[0]["value_json"])["plan_id"] == "raw-authority-frontier:cycle-two"


def test_inspect_conflicts_membership_precondition_evidence(tmp_path: Path) -> None:
raw_id = _seed_byte_proven_browser_head_without_native_id(tmp_path)
with sqlite3.connect(tmp_path / "source.db") as source:
Expand Down