Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
5 changes: 5 additions & 0 deletions docs/reference/local-authority-provider-selection.md
Original file line number Diff line number Diff line change
Expand Up @@ -189,6 +189,11 @@ CLI 使用上面的完整 JSON 和 preview/apply/inspect/bootstrap 命令;App
审核计划;反向迁移须保留新增写入。受支持的旧备份、格式和原回执恢复能力保留。
此路径不代表完整升级、D2 长期资格或发布默认已通过。

Prose-only source maintenance compares the exact canonical JSON bytes of the
existing stable partition view. Only the resume evaluation clock is excluded;
boolean and integer facts remain distinct even where Python object equality
would equate them. This uses the same encoding boundary as capture identity,
without granting a prose writer any Todo mutation or provider fallback.
## Retirement boundaries before changing the release default

SQLite adoption and Python retirement need separate evidence. A canonical Goal
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ def require_prose_state_write_allowed(
from ...rollout_event_log import load_rollout_events, rollout_event_log_path
from ..todos.todo_index import MAX_TODO_INDEX_ROLLOUT_EVENTS_PER_GOAL
from .local_authority_shadow_adapter import todo_partition_projector
from .local_authority_shadow_projection import partition_comparison_view
from .local_authority_shadow_projection import canonical_bytes, partition_comparison_view

try:
goal = find_registry_goal(load_registry(registry_path), goal_id)
Expand All @@ -54,7 +54,9 @@ def require_prose_state_write_allowed(
# two projections were evaluated milliseconds apart.
original_projection = partition_comparison_view(projector(original_text))
planned_projection = partition_comparison_view(projector(planned_text))
if original_projection != planned_projection:
# Python structural equality aliases JSON booleans with integers.
# Use the existing exact-byte identity contract before authorizing IO.
if canonical_bytes(original_projection) != canonical_bytes(planned_projection):
raise ActiveStateAuthorityMutationError(
"prose update would change canonical Todo or handoff state"
)
Expand Down
45 changes: 45 additions & 0 deletions tests/control_plane/test_shadow_writer_boundaries.py
Original file line number Diff line number Diff line change
Expand Up @@ -492,6 +492,51 @@ def project_at_distinct_time(text):
)



@pytest.mark.parametrize("original_value,planned_value", [(True, 1), (False, 0)])
def test_prose_guard_preserves_json_scalar_identity(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
original_value: bool, planned_value: int,
) -> None:
"""A full source projection must retain JSON types even if Python equates them."""
from loopx.control_plane.coordination import local_authority_shadow_adapter as adapter
from loopx.control_plane.coordination.local_authority_shadow_projection import (
todo_partition_projection,
)
from loopx.control_plane.coordination.runtime_shadow_writer_adapter import (
ActiveStateAuthorityMutationError, require_prose_state_write_allowed,
)

registry, state, root = fixture(tmp_path)
add_goal_todo(registry_path=registry, goal_id=GOAL, role="agent", text="Retain result facts.")
original = state.read_text(encoding="utf-8")
planned = original + "\nA narrative observation.\n"
real_projector = adapter.todo_partition_projector

def source_projector(*args, **kwargs):
project = real_projector(*args, **kwargs)

def project_result(text):
projection = project(text)
# Exercise the complete-record adapter contract through the real TS
# assembler. This injected result fact models an adapter output,
# not an assertion that the current Markdown codec emits this shape.
projection["todos"][0]["completion_result"] = {
"facts": [{"value": original_value if text == original else planned_value}],
}
return todo_partition_projection(**projection)

return project_result

monkeypatch.setattr(adapter, "todo_partition_projector", source_projector)
with pytest.raises(ActiveStateAuthorityMutationError, match="would change canonical"):
require_prose_state_write_allowed(
registry_path=registry, runtime_root=root, goal_id=GOAL,
state_path=state, original_text=original, planned_text=planned,
)
assert state.read_text(encoding="utf-8") == original


def test_prose_only_reward_holds_before_index_append_during_maintenance(tmp_path: Path) -> None:
from loopx.feedback import append_human_reward
from loopx.control_plane.coordination.shadow_management import ShadowManagementError, shadow_management_state_path
Expand Down
Loading