mirror of
https://github.com/razzant/ouroboros.git
synced 2026-10-02 19:58:46 +00:00
9c4bf0b5b3
Owner decisions Q8=B (quarantine, NO converter, no compat machinery) and 6.3=B (log-only visibility: no UI counter, no chat notice). - New leaf ouroboros/task_result_schema.py (re-exported by task_results.py, module-size discipline): typed refusal classifier (malformed / unstamped_pre_7_0 / future_schema / invalid_stamp / retired_contract_until_deadline - the f30 ledger residual), writer-side future-refusal guard, lock-guarded byte-preserving quarantine move into task_results/quarantine/, ONE durable task_results_quarantined events row per read/scan batch. - Writers stamp on write: write_task_result, acceptance-state and plan-review merge-writers, owner_hurry projection writer. The cancel-receipt amend-writer (supervisor/terminal_delivery.py) is left byte-identical: it never creates a row and its merge preserves whatever stamp the row carries (the file sits exactly at the 1500-line band edge). - Readers: load_task_result/list_task_results quarantine inadmissible rows on the fail-soft path and raise WITHOUT mutating storage on the strict path (pre-ABI-2 strict message kept stable for malformed rows). - state.json / queue_snapshot.json: stamp-on-write only, form unchanged, readers require no stamp (verbatim Q8=B). - F12 suite tests/test_task_result_schema_quarantine.py: future-refusal, malformed, idempotency, N-1 (live-row stamp-on-write + unstamped state/snapshot reads), rollback (stamp is one additive key), batch-event and no-chat-notice pins; N-1/rollback fixtures shared with the ABI-7 shim suite. - Hand-written test fixtures representing CURRENT-version writers now stamp their rows (9 test files). - ARCHITECTURE.md: same-commit deltas (module bullets, data layout, §11.1 schema_versions row, §11.2 unfrozen-surfaces bullet). Ratchet manifest byte-identical (task_results.py stays in the 1001-1500 band at 1465 lines after the leaf split). (cherry picked from commit b2d2cd554b6a9d2ceb8d39a10378659d20f97996)
86 lines
3 KiB
Python
86 lines
3 KiB
Python
"""Public task results expose the canonical read-only Plan Review v1 projection."""
|
|
|
|
import asyncio
|
|
import copy
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from ouroboros.gateway.tasks import api_task_get
|
|
from ouroboros.outcomes import public_task_result
|
|
from ouroboros.utils import atomic_write_json, read_json_dict
|
|
|
|
|
|
def _legacy(kind: str) -> dict:
|
|
fingerprint = "a" * 64
|
|
state = {
|
|
"schema_version": 1,
|
|
"current_attempt": {},
|
|
"latest_review_fingerprint": "",
|
|
"waves": [],
|
|
}
|
|
if kind == "pending":
|
|
state["waves"] = [{"request_fingerprint": fingerprint, "phase": "collected"}]
|
|
return state
|
|
state["current_attempt"] = {
|
|
"fingerprint": fingerprint,
|
|
"status": "rail_degraded" if kind == "rail_degraded" else (
|
|
"unavailable" if kind == "unavailable" else "open"
|
|
),
|
|
"reason": "deadline" if kind == "rail_degraded" else (
|
|
"reviewer unavailable" if kind == "unavailable" else ""
|
|
),
|
|
}
|
|
state["latest_review_fingerprint"] = fingerprint
|
|
if kind != "unavailable":
|
|
closed = kind == "closed"
|
|
state["waves"] = [{
|
|
"request_fingerprint": fingerprint,
|
|
"phase": "reviewed",
|
|
"review_evidence_status": "integrated",
|
|
"review": {
|
|
"aggregate_signal": "GREEN" if closed else "REVIEW_REQUIRED",
|
|
"closed": closed,
|
|
},
|
|
}]
|
|
return state
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("kind", "status", "outcome", "closed"),
|
|
[
|
|
("open", "open", "REVIEW_REQUIRED", False),
|
|
("pending", "pending", "", False),
|
|
("unavailable", "open", "", False),
|
|
("rail_degraded", "rail_degraded", "REVIEW_REQUIRED", False),
|
|
("closed", "closed", "GREEN", True),
|
|
],
|
|
)
|
|
def test_public_result_adds_legacy_projection_without_mutating_authority(
|
|
kind, status, outcome, closed,
|
|
):
|
|
raw = _legacy(kind)
|
|
before = copy.deepcopy(raw)
|
|
public = public_task_result({"task_id": "legacy", "plan_review_state": raw})
|
|
assert raw == before
|
|
assert public["plan_review_state"]["schema_version"] == 1
|
|
projection = public["plan_review_state"]["legacy_v1_projection"]
|
|
assert (projection["status"], projection["outcome"], projection["closed"]) == (
|
|
status, outcome, closed,
|
|
)
|
|
|
|
|
|
def test_task_detail_projects_legacy_state_without_rewriting_disk(tmp_path):
|
|
task_id = "legacy-detail"
|
|
path = tmp_path / "task_results" / f"{task_id}.json"
|
|
path.parent.mkdir(parents=True)
|
|
stored = {"_schema_version": 1, "task_id": task_id, "status": "running", "plan_review_state": _legacy("open")}
|
|
atomic_write_json(path, stored)
|
|
request = SimpleNamespace(
|
|
path_params={"task_id": task_id},
|
|
app=SimpleNamespace(state=SimpleNamespace(drive_root=tmp_path)),
|
|
)
|
|
payload = json.loads(asyncio.run(api_task_get(request)).body.decode("utf-8"))
|
|
assert payload["plan_review_state"]["legacy_v1_projection"]["outcome"] == "REVIEW_REQUIRED"
|
|
assert read_json_dict(path) == stored
|