Skip to content

fix(storage): preserve logical supersession in census postflight (#3269) #3269

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 2 commits into from
Jul 23, 2026
Merged

fix(storage): preserve logical supersession in census postflight (#3269) #3269

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4e49ed9
fix(storage): preserve logical supersession in census postflight
Sinity Jul 23, 2026
6286dc3
fix(storage): require terminal logical supersession
Sinity Jul 23, 2026
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
28 changes: 24 additions & 4 deletions polylogue/storage/raw_authority.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1320,16 +1320,33 @@ def finalize_raw_authority_census(
SELECT p.input_raw_ids_json
FROM raw_authority_census_plans AS cp
JOIN raw_authority_plans AS p ON p.plan_id = cp.plan_id
WHERE cp.census_id = ? AND cp.selected = 1
WHERE cp.census_id = ?
AND cp.selected = 1
AND cp.outcome_status IN ('executed', 'terminal')
""",
(census_id,),
)
for raw_id in json.loads(str(input_raw_ids_json))
}
selected_logical_keys = {
str(logical_key)
for (logical_keys_json,) in conn.execute(
"""
SELECT p.logical_keys_json
FROM raw_authority_census_plans AS cp
JOIN raw_authority_plans AS p ON p.plan_id = cp.plan_id
WHERE cp.census_id = ?
AND cp.selected = 1
AND cp.outcome_status IN ('executed', 'terminal')
""",
(census_id,),
)
for logical_key in json.loads(str(logical_keys_json))
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
persistent: set[str] = set()
for plan_id, outcome_status, input_raw_ids_json in conn.execute(
for plan_id, outcome_status, input_raw_ids_json, logical_keys_json in conn.execute(
"""
SELECT cp.plan_id, cp.outcome_status, p.input_raw_ids_json
SELECT cp.plan_id, cp.outcome_status, p.input_raw_ids_json, p.logical_keys_json
FROM raw_authority_census_plans AS cp
JOIN raw_authority_plans AS p ON p.plan_id = cp.plan_id
WHERE cp.census_id = ?
Expand All @@ -1338,7 +1355,10 @@ def finalize_raw_authority_census(
(census_id,),
):
plan_inputs = {str(raw_id) for raw_id in json.loads(str(input_raw_ids_json))}
if str(outcome_status) == RawReplayPlanStatus.RETRYABLE.value or plan_inputs.isdisjoint(selected_inputs):
plan_logical_keys = {str(logical_key) for logical_key in json.loads(str(logical_keys_json))}
if str(outcome_status) == RawReplayPlanStatus.RETRYABLE.value or (
plan_inputs.isdisjoint(selected_inputs) and plan_logical_keys.isdisjoint(selected_logical_keys)
):
persistent.add(str(plan_id))
if not persistent.issubset(post_ids):
raise RuntimeError(
Expand Down
168 changes: 168 additions & 0 deletions tests/unit/storage/test_raw_authority_ledger.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -336,6 +336,174 @@ def test_interrupted_census_has_no_partial_plan_visibility_and_retries_once(tmp_
assert finalized.lifecycle_status == "interrupted"


def test_postflight_allows_carried_plan_superseded_by_selected_logical_source(tmp_path: Path) -> None:
"""A selected authority change may retire another raw for its logical source.

Anti-vacuity: treating raw-id disjointness as the only relationship makes
finalization reject this valid postflight because ``retired`` is omitted.
"""
initialize_active_archive_root(tmp_path)
selected = RawReplayPlan(
plan_id="selected",
input_digest="a" * 64,
input_raw_ids=("selected-raw",),
logical_keys=("chatgpt:shared",),
authority_witness=json_document({}),
source_preconditions=json_document({}),
index_preconditions=json_document({}),
)
retired = RawReplayPlan(
plan_id="retired",
input_digest="b" * 64,
input_raw_ids=("retired-raw",),
logical_keys=("chatgpt:shared",),
authority_witness=json_document({}),
source_preconditions=json_document({}),
index_preconditions=json_document({}),
)
independent = RawReplayPlan(
plan_id="independent",
input_digest="c" * 64,
input_raw_ids=("independent-raw",),
logical_keys=("chatgpt:independent",),
authority_witness=json_document({}),
source_preconditions=json_document({}),
index_preconditions=json_document({}),
)
receipt = record_raw_authority_census(
tmp_path,
(selected, retired, independent),
selected_plan_ids={selected.plan_id},
mode="apply",
quiescent=True,
scope={"test": "logical-supersession"},
residual={},
)
record_raw_replay_outcome(
tmp_path,
receipt.census_id,
RawReplayPlanOutcome(
selected.plan_id,
selected.input_raw_ids,
RawReplayPlanStatus.EXECUTED,
"selected authority reached terminal state",
"none",
),
)

finalized = finalize_raw_authority_census(
tmp_path,
receipt.census_id,
post_plans=(independent,),
post_residual={},
)

assert finalized.lifecycle_status == "completed"
assert finalized.post_plan_count == 1


def test_postflight_rejects_missing_independent_carried_plan(tmp_path: Path) -> None:
"""Logical-source supersession must not weaken independent-plan protection."""
initialize_active_archive_root(tmp_path)
selected = RawReplayPlan(
plan_id="selected",
input_digest="a" * 64,
input_raw_ids=("selected-raw",),
logical_keys=("chatgpt:selected",),
authority_witness=json_document({}),
source_preconditions=json_document({}),
index_preconditions=json_document({}),
)
independent = RawReplayPlan(
plan_id="independent",
input_digest="c" * 64,
input_raw_ids=("independent-raw",),
logical_keys=("chatgpt:independent",),
authority_witness=json_document({}),
source_preconditions=json_document({}),
index_preconditions=json_document({}),
)
receipt = record_raw_authority_census(
tmp_path,
(selected, independent),
selected_plan_ids={selected.plan_id},
mode="apply",
quiescent=True,
scope={"test": "independent-preservation"},
residual={},
)
record_raw_replay_outcome(
tmp_path,
receipt.census_id,
RawReplayPlanOutcome(
selected.plan_id,
selected.input_raw_ids,
RawReplayPlanStatus.EXECUTED,
"selected authority reached terminal state",
"none",
),
)

with pytest.raises(RuntimeError, match="postflight changed a retryable/carried-forward plan"):
finalize_raw_authority_census(
tmp_path,
receipt.census_id,
post_plans=(),
post_residual={},
)


def test_postflight_rejects_carried_plan_shared_with_retryable_selection(tmp_path: Path) -> None:
"""Retryable work cannot retire a different raw for the same logical source."""
initialize_active_archive_root(tmp_path)
retryable = RawReplayPlan(
plan_id="retryable",
input_digest="a" * 64,
input_raw_ids=("retryable-raw",),
logical_keys=("chatgpt:shared",),
authority_witness=json_document({}),
source_preconditions=json_document({}),
index_preconditions=json_document({}),
)
carried = RawReplayPlan(
plan_id="carried",
input_digest="b" * 64,
input_raw_ids=("carried-raw",),
logical_keys=("chatgpt:shared",),
authority_witness=json_document({}),
source_preconditions=json_document({}),
index_preconditions=json_document({}),
)
receipt = record_raw_authority_census(
tmp_path,
(retryable, carried),
selected_plan_ids={retryable.plan_id},
mode="apply",
quiescent=True,
scope={"test": "retryable-does-not-supersede"},
residual={},
)
record_raw_replay_outcome(
tmp_path,
receipt.census_id,
RawReplayPlanOutcome(
retryable.plan_id,
retryable.input_raw_ids,
RawReplayPlanStatus.RETRYABLE,
"selected authority remains executable",
"retry after the current writer pass",
),
)

with pytest.raises(RuntimeError, match="postflight changed a retryable/carried-forward plan"):
finalize_raw_authority_census(
tmp_path,
receipt.census_id,
post_plans=(),
post_residual={},
)


def test_global_census_quiesces_moved_component_before_any_plan_is_published(tmp_path: Path) -> None:
initialize_active_archive_root(tmp_path)
first = _write_codex_raw(
Expand Down