From ea0074966c5178dc11a891ba77ce37fb211f7408 Mon Sep 17 00:00:00 2001 From: Aram Aljanadi Date: Tue, 27 Oct 2026 21:31:33 +0100 Subject: [PATCH 1/5] Add Task IR P1 compiler and runner mode Four typed operations compile into existing machinery: reads become one agent plan, route.patch preview/apply rides the reviewed seal flow. No new los verbs, approval format, or transaction engine. --- pyproject.toml | 1 + .../components/operation-pipeline.md | 1 + tools/agent_run.py | 34 + tools/learning_os/app/task_ir.py | 613 ++++++++++++++++++ 4 files changed, 649 insertions(+) create mode 100644 tools/learning_os/app/task_ir.py diff --git a/pyproject.toml b/pyproject.toml index c8e8e32..eaff594 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -77,6 +77,7 @@ files = [ "tools/learning_os/app/agent_plans.py", "tools/learning_os/adapters/agent_runner.py", "tools/learning_os/app/agent_presentation.py", + "tools/learning_os/app/task_ir.py", ] # Linting (engineering review 2026-08-18, item 12; added 2026-08-21). diff --git a/system/developer/components/operation-pipeline.md b/system/developer/components/operation-pipeline.md index 3115358..97c0697 100644 --- a/system/developer/components/operation-pipeline.md +++ b/system/developer/components/operation-pipeline.md @@ -28,6 +28,7 @@ Admit, lock, dispatch and settle one operation through the shared application bo - `tools/learning_os/app/agent_plans.py` - `tools/learning_os/app/agent_presentation.py` - `tools/learning_os/app/agent_reporting.py` +- `tools/learning_os/app/task_ir.py` - `tools/learning_os/app/projection_readiness.py` - `tools/learning_os/app/read_session.py` diff --git a/tools/agent_run.py b/tools/agent_run.py index 7b95503..98ee30a 100644 --- a/tools/agent_run.py +++ b/tools/agent_run.py @@ -31,13 +31,47 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("--report-dir", default=None, help="directory for the full report artifact (bounded stdout)") parser.add_argument("--keep-report", action="store_true", help="exclude this run from report retention") + parser.add_argument("--task", default=None, + help="Task IR P1 spec JSON file (experimental; alternative to a plan)") + parser.add_argument("--task-phase", choices=("preview", "apply"), default="preview", + help="Task IR phase: preview seals the reviewed check only; apply submits it once") + parser.add_argument("--evidence-dir", default=None, + help="Task IR run-evidence directory (outside the data root)") return parser +def _run_task(args: argparse.Namespace, root: Path) -> int: + from learning_os.app import task_ir + + if args.plan or args.expand_report or args.step: + raise PlanStepError("choose a plan, --expand-report, or --task") + if not args.evidence_dir: + raise PlanStepError("--task requires --evidence-dir outside the data root") + if Path(args.task).stat().st_size > 262144: + raise PlanStepError("task spec exceeds 256 KiB") + spec = task_ir.parse_spec(json.loads( + Path(args.task).read_text(encoding="utf-8"), + parse_float=_finite_number, parse_constant=_finite_number)) + evidence_dir = Path(args.evidence_dir) + if args.task_phase == "preview": + report = task_ir.preview_task( + root, spec, evidence_dir, + run_reads=lambda plan, data_root: agent_runner.run_plan( + plan, data_root, report_dir=args.report_dir, + keep_report=args.keep_report)) + else: + report = task_ir.apply_task(root, spec, evidence_dir) + print(encode_json(report)) + return {"ok": 0, "needs_review": 0, "committed": 0, "unchanged": 0, + "refused": 2, "partial": 3, "uncertain": 3}.get(str(report["status"]), 3) + + def main() -> int: args = build_parser().parse_args() root = Path(args.root).resolve() if args.root else TOOLS.parent try: + if args.task: + return _run_task(args, root) if args.expand_report: from learning_os.app.agent_reporting import expand_report if args.plan: diff --git a/tools/learning_os/app/task_ir.py b/tools/learning_os/app/task_ir.py new file mode 100644 index 0000000..abfc38e --- /dev/null +++ b/tools/learning_os/app/task_ir.py @@ -0,0 +1,613 @@ +"""Task IR P1: four typed operations compiled into existing Core machinery. + +Experimental and narrow by design. A task spec names typed operations over +existing canonical IDs; this module validates the spec, compiles its reads +into one agent plan for the existing runner, and drives the single writable +operation (``route.patch``) through the existing reviewed seal flow +(``seal_envelope.py --guards auto --check-out`` at preview, then +``--guards-from ... --submit`` at apply). No new approval format, no new +transaction engine, no new session entity, and no ``los`` verbs are added: +the runner mode in ``tools/agent_run.py`` is the only new surface. + +Operation mapping (effects always come from the registry classification of +the underlying operation, never from a name): + +- ``context.inspect`` -> ``inspect`` (``project.inspect``, read): one known + unit, stage, route, or record ID. +- ``material.find`` -> ``plan-edit-context`` (``plan.edit-context``, read): + the typed routes of one known unit, or of one stage of it. Lexical + explanation search (``material-context``) stays a direct call in P1. +- ``session.propose`` -> ``runtime-session`` (``runtime.session``, read): + the requirement ID is derived with + :func:`learning_os.derive.learning_runtime.requirement_id_for`; no + requirement store is invented. +- ``route.patch`` -> ``route.patch`` (canonical-write): preview plus one + guarded canonical write through the Gateway. + +Typed references only: an operation input may name an earlier operation's +documented output through ``{"from_operation": ..., "field": ...}``. +String expression languages (``$steps[...]`` and friends) are refused. + +Idempotency: the spec digest is the SHA-256 of the runner's canonical form +(:func:`learning_os.app.agent_reporting.encode`) of the normalized spec. +Reusing a ``request_id`` with a different digest is an error. The write +operation's idempotency key and gateway request ID are derived once at +preview, persisted in the run evidence keyed by ``(request_id, +operation_id)``, and looked up on every retry: apply resubmits through the +saved check record and the gateway's kept envelope, never resealing against +a fresh snapshot, and an uncertain dispatch is never redispatched. +""" + +from __future__ import annotations + +import hashlib +import json +import re +import subprocess +import sys +from collections.abc import Callable +from dataclasses import dataclass, field +from pathlib import Path +from typing import cast + +from learning_os.app import handlers as _handlers # noqa: F401 (registration) +from learning_os.app.agent_reporting import encode +from learning_os.app.registry import REGISTRY, OperationEffects +from learning_os.derive.learning_runtime import requirement_id_for +from learning_os.domain.material_refs import PATCH_FIELDS + +TOOLS = Path(__file__).resolve().parents[2] + +_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]*$") +_OP_ID = re.compile(r"^[a-z][a-z0-9-]{0,63}$") + +#: The one writable workflow P1 supports: exactly one route.patch per spec. +MAX_OPERATIONS = 16 + +#: Documented typed outputs per operation (the only fields a reference may +#: name). Reads expose identifiers and the snapshot they were read at; the +#: write exposes its receipt-backed evidence. +OUTPUTS: dict[str, tuple[str, ...]] = { + "context.inspect": ("record_id", "snapshot_id"), + "material.find": ("route_id", "route_ids", "snapshot_id"), + "session.propose": ("requirement_id", "snapshot_id"), + "route.patch": ("route_id", "receipt_path", "transaction_id"), +} + +#: IR operation -> underlying registry operation whose declared effects +#: classify the IR operation. +UNDERLYING: dict[str, str] = { + "context.inspect": "project.inspect", + "material.find": "plan.edit-context", + "session.propose": "runtime.session", + "route.patch": "route.patch", +} + + +class TaskSpecError(ValueError): + """A task spec, reference, or persisted run state is invalid.""" + + def __init__(self, message: str, code: str = "TASK_IR_INVALID") -> None: + super().__init__(message) + self.code = code + + +@dataclass(frozen=True) +class RouteRef: + """A route identity: one literal canonical ID or one typed reference.""" + + literal: str | None = None + from_operation: str | None = None + + +@dataclass(frozen=True) +class Operation: + id: str + op: str + unit_id: str | None = None + stage_id: str | None = None + route_id: str | None = None + record_id: str | None = None + route_ref: RouteRef | None = None + set_fields: dict[str, str] = field(default_factory=dict) + claim_evidence: dict[str, object] | None = None + context: dict[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True) +class TaskSpec: + request_id: str + session_id: str + operations: tuple[Operation, ...] + normalized: dict[str, object] + + +@dataclass(frozen=True) +class CompiledTask: + plan: dict[str, object] + write: Operation | None + read_ids: tuple[str, ...] + + +def _fail(message: str, code: str = "TASK_IR_INVALID") -> TaskSpecError: + return TaskSpecError(message, code) + + +def _text(raw: dict[str, object], name: str, op_id: str, *, + required: bool = False) -> str | None: + value = raw.get(name) + if value is None: + if required: + raise _fail(f"operation {op_id}: {name} is required") + return None + if not isinstance(value, str) or not value.strip(): + raise _fail(f"operation {op_id}: {name} must be a nonempty string") + return value + + +def _route_ref(raw: dict[str, object], op_id: str, + earlier: dict[str, Operation]) -> RouteRef: + has_literal = "route_id" in raw + has_ref = "route_ref" in raw + if has_literal == has_ref: + raise _fail(f"operation {op_id}: name exactly one of route_id or route_ref") + if has_literal: + value = raw["route_id"] + if not isinstance(value, str) or not value.strip(): + raise _fail(f"operation {op_id}: route_id must be a nonempty string") + return RouteRef(literal=value) + ref = raw["route_ref"] + if not isinstance(ref, dict) or set(ref) != {"from_operation", "field"}: + raise _fail(f"operation {op_id}: route_ref must name from_operation and field") + source, field_name = ref["from_operation"], ref["field"] + if not isinstance(source, str) or source not in earlier: + raise _fail(f"operation {op_id}: route_ref names unknown or later operation {source!r}", + "TASK_IR_BAD_REFERENCE") + if field_name != "route_id" or field_name not in OUTPUTS[earlier[source].op]: + raise _fail(f"operation {op_id}: operation {source} exposes no route_id output", + "TASK_IR_BAD_REFERENCE") + return RouteRef(from_operation=source) + + +def _operation(raw: object, earlier: dict[str, Operation]) -> Operation: + if not isinstance(raw, dict): + raise _fail("each operation must be an object") + op_id = raw.get("id") + if not isinstance(op_id, str) or not _OP_ID.fullmatch(op_id): + raise _fail("operation id must match ^[a-z][a-z0-9-]{0,63}$") + if op_id in earlier: + raise _fail(f"duplicate operation id: {op_id}") + op = raw.get("op") + if op not in OUTPUTS: + raise _fail(f"operation {op_id}: unknown op {op!r}") + allowed = {"id", "op"} + if op == "context.inspect": + allowed |= {"record_id", "unit_id", "stage_id", "route_id"} + record = raw.get("record_id") or raw.get("route_id") or raw.get("stage_id") or raw.get("unit_id") + if not isinstance(record, str) or not record.strip(): + raise _fail(f"operation {op_id}: context.inspect needs a known unit, stage, route, or record id") + return Operation(id=op_id, op=op, record_id=record) + if op == "material.find": + allowed |= {"unit_id", "stage_id"} + return Operation(id=op_id, op=op, + unit_id=_text(raw, "unit_id", op_id, required=True), + stage_id=_text(raw, "stage_id", op_id)) + if op == "session.propose": + allowed |= {"unit_id", "stage_id", "context"} + context = raw.get("context", {}) + if not isinstance(context, dict): + raise _fail(f"operation {op_id}: context must be an object") + return Operation(id=op_id, op=op, + unit_id=_text(raw, "unit_id", op_id, required=True), + stage_id=_text(raw, "stage_id", op_id, required=True), + context=dict(context)) + allowed |= {"unit_id", "route_id", "route_ref", "set_fields", "claim_evidence"} + route_ref = _route_ref(raw, op_id, earlier) + set_fields = raw.get("set_fields") + if not isinstance(set_fields, dict) or not set_fields: + raise _fail(f"operation {op_id}: set_fields must be a nonempty object") + for key, value in set_fields.items(): + if key not in PATCH_FIELDS: + raise _fail(f"operation {op_id}: set_fields field {key!r} is not a declared material field") + if not isinstance(value, str) or not value.strip(): + raise _fail(f"operation {op_id}: set_fields.{key} must be nonempty text") + evidence = raw.get("claim_evidence") + if evidence is not None and not isinstance(evidence, dict): + raise _fail(f"operation {op_id}: claim_evidence must be an object") + extra = set(raw) - allowed + if extra: + raise _fail(f"operation {op_id}: undeclared field(s): {', '.join(sorted(extra))}") + return Operation(id=op_id, op=op, + unit_id=_text(raw, "unit_id", op_id, required=True), + route_ref=route_ref, set_fields=dict(set_fields), + claim_evidence=dict(evidence) if evidence is not None else None) + + +def _reject_expressions(value: object) -> None: + if isinstance(value, str) and ("$" in value or value.startswith("{")): + raise _fail("string expressions are not typed references; use route_ref", "TASK_IR_BAD_REFERENCE") + if isinstance(value, dict): + for item in value.values(): + _reject_expressions(item) + if isinstance(value, list): + for item in value: + _reject_expressions(item) + + +def parse_spec(raw: object) -> TaskSpec: + """Validate one task spec; the only admission path for task IR input.""" + if not isinstance(raw, dict): + raise _fail("task spec must be a JSON object") + extra = set(raw) - {"schema_version", "request_id", "session_id", "operations"} + if extra: + raise _fail(f"undeclared spec field(s): {', '.join(sorted(extra))}") + if raw.get("schema_version") != 1: + raise _fail("task spec schema_version must be 1") + request_id = raw.get("request_id") + if not isinstance(request_id, str) or not _ID.fullmatch(request_id) or len(request_id) > 128: + raise _fail("request_id must match ^[A-Za-z0-9][A-Za-z0-9._:-]*$ (max 128)") + session_id = raw.get("session_id", f"task-{request_id}") + if not isinstance(session_id, str) or not session_id.strip() or len(session_id) > 128: + raise _fail("session_id must be a nonempty string (max 128)") + raw_ops = raw.get("operations") + if not isinstance(raw_ops, list) or not 1 <= len(raw_ops) <= MAX_OPERATIONS: + raise _fail(f"operations must hold 1..{MAX_OPERATIONS} operations") + _reject_expressions(raw_ops) + operations: dict[str, Operation] = {} + ordered: list[Operation] = [] + for item in raw_ops: + operation = _operation(item, operations) + operations[operation.id] = operation + ordered.append(operation) + writes = [item for item in ordered if item.op == "route.patch"] + if len(writes) > 1: + raise _fail("P1 admits exactly one canonical transaction: at most one route.patch") + for operation in ordered: + entry = REGISTRY.get(UNDERLYING[operation.op]) + if entry is None: + raise _fail(f"underlying operation is not registered: {UNDERLYING[operation.op]}") + expected = (OperationEffects.CANONICAL_WRITE if operation.op == "route.patch" + else OperationEffects.READ) + if entry.effects != expected: + raise _fail(f"{UNDERLYING[operation.op]} is classified {entry.effects}, not {expected}; refusing to infer effects") + normalized: dict[str, object] = { + "schema_version": 1, "request_id": request_id, "session_id": session_id, + "operations": json.loads(encode(raw_ops)), + } + return TaskSpec(request_id=request_id, session_id=session_id, + operations=tuple(ordered), normalized=normalized) + + +def spec_digest(spec: TaskSpec) -> str: + """SHA-256 of the runner's canonical form of the normalized spec.""" + return "sha256:" + hashlib.sha256(encode(spec.normalized).encode("utf-8")).hexdigest() + + +def operation_key(request_id: str, operation_id: str) -> str: + """The write key derived once at preview for (request_id, operation_id).""" + digest = hashlib.sha256(encode([request_id, operation_id]).encode("utf-8")).hexdigest() + return f"task-{digest[:40]}" + + +def compile_spec(spec: TaskSpec) -> CompiledTask: + """Compile reads into one agent plan; the write stays a seal intent. + + Deterministic: the same spec compiles to the same plan bytes. The write + is not a plan step because the runner submits only presealed envelopes; + P1 keeps sealing in the reviewed preview/apply flow instead of + inventing plan syntax for it. + """ + steps: list[dict[str, object]] = [] + write: Operation | None = None + for operation in spec.operations: + if operation.op == "route.patch": + write = operation + continue + depends = ([operation.route_ref.from_operation] if operation.route_ref + and operation.route_ref.from_operation else []) + if operation.op == "context.inspect": + steps.append({"id": operation.id, "argv": ["inspect", str(operation.record_id)], + "summary": {"fields": ["id", "type"], "max_items": 5, "max_bytes": 2000}}) + elif operation.op == "material.find": + argv = ["plan-edit-context", str(operation.unit_id)] + fields = ["snapshot_id", "routes"] + if operation.stage_id: + argv += ["--stage-id", operation.stage_id] + fields = ["snapshot_id", "stage.resources"] + steps.append({"id": operation.id, "argv": argv, "depends_on": depends, + "summary": {"fields": fields, "max_items": 20, "max_bytes": 8000}}) + else: + requirement = requirement_id_for(str(operation.unit_id), str(operation.stage_id)) + steps.append({"id": operation.id, + "argv": ["runtime-session", "--requirement", requirement, + "--context-json", encode(operation.context)], + "summary": {"fields": ["snapshot_id", "contract"], + "max_items": 5, "max_bytes": 2000}}) + plan: dict[str, object] = {"schema_version": 2, "session_id": spec.session_id, + "on_error": "stop", "steps": steps} + return CompiledTask(plan=plan, write=write, + read_ids=tuple(op.id for op in spec.operations + if op.op != "route.patch")) + + +def resolve_route(compiled: CompiledTask, outputs: dict[str, dict[str, object]]) -> str: + """Bind the write's route from a literal ID or an earlier find output.""" + write = compiled.write + if write is None or write.route_ref is None: + raise _fail("compiled task has no route.patch to bind") + if write.route_ref.literal is not None: + return write.route_ref.literal + source = outputs.get(str(write.route_ref.from_operation), {}) + route_id = source.get("route_id") + if not isinstance(route_id, str) or not route_id: + raise _fail("material.find did not resolve to exactly one route; refusing to guess", + "TASK_IR_UNRESOLVED_ROUTE") + return route_id + + +def write_payload(write: Operation, route_id: str) -> dict[str, object]: + payload: dict[str, object] = {"unit_id": write.unit_id, "route_id": route_id, + "changes": dict(write.set_fields)} + if write.claim_evidence is not None: + payload["claim_evidence"] = write.claim_evidence + return payload + + +ReadRunner = Callable[[dict[str, object], Path], dict[str, object]] +SealCheck = Callable[[Path, dict[str, object], str, Path], tuple[int, str, str]] +SealSubmit = Callable[[Path, Path, dict[str, object], str, str], tuple[int, str, str]] + + +def _run_seal(argv: list[str]) -> tuple[int, str, str]: + proc = subprocess.run(argv, capture_output=True, text=True, timeout=600) + return proc.returncode, proc.stdout, proc.stderr + + +def default_seal_check(root: Path, payload: dict[str, object], key: str, + record_path: Path) -> tuple[int, str, str]: + return _run_seal([sys.executable, str(TOOLS / "seal_envelope.py"), "--root", str(root), + "--capability", "route.patch", "--payload", encode(payload), + "--key", key, "--guards", "auto", "--check-out", str(record_path)]) + + +def default_seal_submit(root: Path, record_path: Path, payload: dict[str, object], + key: str, request_id: str) -> tuple[int, str, str]: + return _run_seal([sys.executable, str(TOOLS / "seal_envelope.py"), "--root", str(root), + "--capability", "route.patch", "--payload", encode(payload), + "--key", key, "--request-id", request_id, + "--guards-from", str(record_path), "--submit"]) + + +def _evidence_path(evidence_dir: Path, request_id: str) -> Path: + name = hashlib.sha256(request_id.encode("utf-8")).hexdigest()[:32] + return evidence_dir / f"task-{name}.json" + + +def _save_evidence(path: Path, evidence: dict[str, object]) -> None: + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + tmp = path.with_suffix(".tmp") + tmp.write_text(encode(evidence) + "\n", encoding="utf-8") + tmp.replace(path) + + +def _load_evidence(path: Path, spec: TaskSpec) -> dict[str, object] | None: + if not path.is_file(): + return None + try: + evidence = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise _fail(f"run evidence is unreadable: {path}", "TASK_IR_EVIDENCE") from exc + if not isinstance(evidence, dict) or evidence.get("request_id") != spec.request_id: + raise _fail("run evidence belongs to another request", "TASK_IR_EVIDENCE") + if evidence.get("spec_digest") != spec_digest(spec): + raise _fail("request_id was already used with a different spec", "TASK_IR_REQUEST_CONFLICT") + return evidence + + +def _outputs_from_report(spec: TaskSpec, report: dict[str, object]) -> dict[str, dict[str, object]]: + raw_steps = report.get("steps") + rows: dict[str, dict[str, object]] = {} + if isinstance(raw_steps, list): + for item in raw_steps: + if isinstance(item, dict): + rows[str(item.get("id"))] = cast(dict[str, object], item) + outputs: dict[str, dict[str, object]] = {} + for operation in spec.operations: + if operation.op == "route.patch": + continue + row: dict[str, object] = rows.get(operation.id, {}) + raw_output = row.get("output") + output: dict[str, object] = raw_output if isinstance(raw_output, dict) else {} + if operation.op == "context.inspect": + outputs[operation.id] = {"record_id": operation.record_id, + "snapshot_id": output.get("snapshot_id"), + "status": row.get("status")} + elif operation.op == "material.find": + route_ids: list[str] = [] + stage = output.get("stage") + if isinstance(stage, dict) and isinstance(stage.get("resources"), list): + route_ids = [str(item["route_id"]) for item in stage["resources"] + if isinstance(item, dict) and item.get("route_id")] + raw_routes = output.get("routes") + if isinstance(raw_routes, list): + route_ids = [str(item["id"]) for item in raw_routes + if isinstance(item, dict) and item.get("id")] + route_ids = list(dict.fromkeys(route_ids)) + outputs[operation.id] = { + "route_ids": route_ids, + "route_id": route_ids[0] if len(route_ids) == 1 else None, + "snapshot_id": output.get("snapshot_id"), "status": row.get("status")} + else: + outputs[operation.id] = { + "requirement_id": requirement_id_for(str(operation.unit_id), str(operation.stage_id)), + "snapshot_id": output.get("snapshot_id"), "status": row.get("status")} + return outputs + + +def _record_changes(record: dict[str, object]) -> dict[str, object] | None: + """The previewed per-field changes from a seal check record, if carried.""" + for value in record.values(): + if isinstance(value, dict): + changes = value.get("changes") + if isinstance(changes, dict): + return changes + nested = _record_changes(value) + if nested is not None: + return nested + return None + + +def preview_task(root: Path, spec: TaskSpec, evidence_dir: Path, *, + run_reads: ReadRunner, + seal_check: SealCheck = default_seal_check) -> dict[str, object]: + """Run reads and the reviewed check; zero canonical writes, zero receipts.""" + if evidence_dir.resolve().is_relative_to(root.resolve()): + raise _fail("evidence directory must live outside the data root", "TASK_IR_EVIDENCE") + evidence_dir.mkdir(parents=True, exist_ok=True, mode=0o700) + path = _evidence_path(evidence_dir, spec.request_id) + prior = _load_evidence(path, spec) + if prior is not None and prior.get("status") in ("committed", "uncertain"): + return _report(spec, prior, resumed=True) + compiled = compile_spec(spec) + empty_report: dict[str, object] = {"steps": [], "totals": {"ok": 0, "steps": 0}} + report = run_reads(compiled.plan, root) if compiled.read_ids else empty_report + outputs = _outputs_from_report(spec, report) + operations: list[dict[str, object]] = [] + for operation in spec.operations: + if operation.op == "route.patch": + continue + status = outputs[operation.id].get("status") + operations.append({"id": operation.id, "op": operation.op, + "status": "ok" if status == "ok" else "refused", + "outputs": {k: v for k, v in outputs[operation.id].items() + if k != "status"}}) + evidence: dict[str, object] = { + "request_id": spec.request_id, "spec_digest": spec_digest(spec), + "status": "previewed", "operations": operations, "outputs": outputs, + } + if compiled.write is not None: + write = compiled.write + failed_read = next((row for row in operations if row["status"] != "ok"), None) + if failed_read is not None: + operations.append({"id": write.id, "op": write.op, "status": "refused", + "outputs": {}, "error": "a read it depends on did not complete"}) + evidence["status"] = "partial" + else: + try: + route_id = resolve_route(compiled, outputs) + except TaskSpecError as exc: + operations.append({"id": write.id, "op": write.op, "status": "refused", + "outputs": {}, "error": str(exc), "code": exc.code}) + evidence["status"] = "partial" + else: + payload = write_payload(write, route_id) + key = operation_key(spec.request_id, write.id) + record_path = evidence_dir / f"task-{spec.request_id}-{write.id}.check.json" + code, _out, err = seal_check(root, payload, key, record_path) + if code != 0 and "no material changes to apply" in err: + # The existing machinery itself refuses a no-change patch + # before any write; that refusal is the unchanged proof. + evidence.update({"key": key, "payload": payload, "route_id": route_id}) + operations.append({"id": write.id, "op": write.op, + "status": "unchanged", + "outputs": {"route_id": route_id}}) + evidence["status"] = "unchanged" + elif code != 0 or not record_path.is_file(): + operations.append({"id": write.id, "op": write.op, "status": "refused", + "outputs": {"route_id": route_id}, + "error": (err.strip().splitlines() or ["preview check failed"])[-1]}) + evidence["status"] = "partial" + else: + record = json.loads(record_path.read_text(encoding="utf-8")) + changes = _record_changes(record) + unchanged = changes is not None and not changes + evidence.update({"key": key, "gateway_request_id": f"request-{key}", + "payload": payload, "check_record": str(record_path), + "route_id": route_id}) + operations.append({"id": write.id, "op": write.op, + "status": "unchanged" if unchanged else "needs_review", + "outputs": {"route_id": route_id}}) + evidence["status"] = "unchanged" if unchanged else "needs_review" + _save_evidence(path, evidence) + return _report(spec, evidence) + + +def _committed_receipt(root: Path, key: str) -> dict[str, object] | None: + from learning_os.store.evidence import load_idempotency_entries + + entry = load_idempotency_entries(root).get(key) + if entry is None: + return None + return {"transaction_id": entry.get("transaction_id"), + "receipt_path": entry.get("receipt_path")} + + +def apply_task(root: Path, spec: TaskSpec, evidence_dir: Path, *, + seal_submit: SealSubmit = default_seal_submit) -> dict[str, object]: + """Apply the previewed write once; retries look up, never redispatch blind.""" + if evidence_dir.resolve().is_relative_to(root.resolve()): + raise _fail("evidence directory must live outside the data root", "TASK_IR_EVIDENCE") + path = _evidence_path(evidence_dir, spec.request_id) + evidence = _load_evidence(path, spec) + if evidence is None: + raise _fail("no preview evidence for this request; run preview first", "TASK_IR_NO_PREVIEW") + if evidence.get("status") in ("committed", "uncertain", "unchanged"): + return _report(spec, evidence, resumed=True) + if evidence.get("status") != "needs_review": + return _report(spec, evidence, resumed=True) + compiled = compile_spec(spec) + assert compiled.write is not None # needs_review is only set with a write + key = str(evidence["key"]) + record_path = Path(str(evidence["check_record"])) + payload = evidence["payload"] + if not isinstance(payload, dict) or not record_path.is_file(): + raise _fail("preview evidence is incomplete", "TASK_IR_EVIDENCE") + evidence["status"] = "dispatched" + _save_evidence(path, evidence) + + def settle(status: str, **extra: object) -> dict[str, object]: + raw_operations = evidence.get("operations") + operations: list[object] = [dict(row) if isinstance(row, dict) else row + for row in raw_operations] \ + if isinstance(raw_operations, list) else [] + for row in operations: + if isinstance(row, dict) and row.get("op") == "route.patch": + row["status"] = status + evidence.update({"status": status, "operations": operations, **extra}) + _save_evidence(path, evidence) + return _report(spec, evidence) + try: + code, out, err = seal_submit(root, record_path, payload, key, + str(evidence["gateway_request_id"])) + except Exception as exc: # dispatch may have crossed the commit boundary + return settle("uncertain", error=f"dispatch interrupted: {exc}") + if code == 0: + result = json.loads(out) if out.strip() else {} + receipt = {"transaction_id": result.get("transaction_id"), + "receipt_path": result.get("receipt_path"), + "replayed": result.get("replayed") is True} + if not receipt["receipt_path"] or not (root / str(receipt["receipt_path"])).is_file(): + return settle("uncertain", + error="gateway reported success without a receipt on disk") + return settle("committed", receipt=receipt) + recovered = _committed_receipt(root, key) + if recovered is not None: + return settle("committed", receipt=recovered) + detail = (err.strip().splitlines() or [out.strip() or "gateway refused the write"])[-1] + return settle("refused", error=detail) + + +def _report(spec: TaskSpec, evidence: dict[str, object], *, + resumed: bool = False) -> dict[str, object]: + operations = evidence.get("operations", []) + status = str(evidence.get("status", "refused")) + report: dict[str, object] = { + "schema_version": 1, "kind": "task-ir-p1", "request_id": spec.request_id, + "spec_digest": evidence.get("spec_digest"), "status": status, + "operations": operations, "resumed": resumed, + } + if evidence.get("receipt") is not None: + report["receipt"] = evidence["receipt"] + if evidence.get("error") is not None: + report["error"] = evidence["error"] + return report From 3541789e5b016ce6503d17a6fa0f9e8e567e5a95 Mon Sep 17 00:00:00 2001 From: Aram Aljanadi Date: Tue, 27 Oct 2026 21:31:41 +0100 Subject: [PATCH 2/5] Add durable Task IR P1 tests and register them Covers the one writable workflow end to end on disposable fixtures, plus rejection, purity, stale, unchanged, and uncertain-dispatch contracts; registers the file in its group and shard leg. --- .github/shard-manifest.json | 1 + tests/group_map.py | 1 + tests/test_task_ir.py | 223 ++++++++++++++++++++++++++++++++++++ 3 files changed, 225 insertions(+) create mode 100644 tests/test_task_ir.py diff --git a/.github/shard-manifest.json b/.github/shard-manifest.json index fb0f4f9..4705236 100644 --- a/.github/shard-manifest.json +++ b/.github/shard-manifest.json @@ -67,6 +67,7 @@ "rest-1": [ "tests/test_agent_efficiency.py", "tests/test_agent_run.py", + "tests/test_task_ir.py", "tests/test_arch_error_details.py", "tests/test_arch_plumbing.py", "tests/test_arch_private_names.py", diff --git a/tests/group_map.py b/tests/group_map.py index bf55c7a..9a354cf 100644 --- a/tests/group_map.py +++ b/tests/group_map.py @@ -129,6 +129,7 @@ "test_agent_efficiency.py", "test_agent_reporting.py", "test_agent_run.py", + "test_task_ir.py", "test_projection_readiness.py", "test_read_session.py", "test_review_consolidation.py", diff --git a/tests/test_task_ir.py b/tests/test_task_ir.py new file mode 100644 index 0000000..982b702 --- /dev/null +++ b/tests/test_task_ir.py @@ -0,0 +1,223 @@ +"""Task IR P1: typed compilation, reviewed preview, one guarded write.""" + +from __future__ import annotations + +import copy +import json +import subprocess +import sys +from pathlib import Path + +import pytest +from repo_builders import material_fixture, run_los + +from learning_os.adapters import agent_runner +from learning_os.app import task_ir +from learning_os.app.agent_reporting import encode +from learning_os.store.fingerprint import canonical_fingerprint +from learning_os.store.loader import load_repo + +ROOT = Path(__file__).resolve().parents[1] + + +def _spec(**overrides): + spec = {"schema_version": 1, "request_id": "task-test-1", "operations": [ + {"id": "inspect-unit", "op": "context.inspect", "unit_id": "unit-demo-l01"}, + {"id": "find-material", "op": "material.find", "unit_id": "unit-demo-l01", + "stage_id": "stage-demo"}, + {"id": "patch-route", "op": "route.patch", "unit_id": "unit-demo-l01", + "route_ref": {"from_operation": "find-material", "field": "route_id"}, + "set_fields": {"angle": "A task-ir angle"}}]} + spec.update(overrides) + return spec + + +def _run_reads(plan, root): + return agent_runner.run_plan(plan, root) + + +def _receipts(root: Path) -> list[str]: + tx = root / "operations/transactions" + return sorted(p.name for p in tx.glob("transaction-*.yaml")) if tx.exists() else [] + + +@pytest.mark.parametrize("mutate", [ + lambda s: s["operations"][0].update(op="context.delete"), + lambda s: s["operations"][2].update(set_fields={"source_id": "x"}), + lambda s: s["operations"][2].update(set_fields={"angle": " "}), + lambda s: s["operations"][2].update(route_ref={"from_operation": "patch-route", "field": "route_id"}), + lambda s: s["operations"][2].update(route_ref={"from_operation": "find-material", "field": "snapshot_id"}), + lambda s: s["operations"][2].update(surprise=True), + lambda s: s.update(schema_version=2), + lambda s: s["operations"].append(dict(s["operations"][2], id="patch-two")), +]) +def test_spec_rejection(mutate): + spec = _spec() + mutate(spec) + with pytest.raises(task_ir.TaskSpecError): + task_ir.parse_spec(spec) + + +def test_expression_references_are_refused(): + spec = _spec() + spec["operations"][2] = {"id": "patch-route", "op": "route.patch", + "unit_id": "unit-demo-l01", + "route_id": "$steps.find-material.route_id", + "set_fields": {"angle": "x"}} + with pytest.raises(task_ir.TaskSpecError): + task_ir.parse_spec(spec) + + +def test_compilation_is_deterministic_and_derives_requirement(): + spec = task_ir.parse_spec(_spec()) + assert encode(task_ir.compile_spec(spec).plan) == encode(task_ir.compile_spec(spec).plan) + propose = task_ir.parse_spec({"schema_version": 1, "request_id": "task-test-2", + "operations": [{"id": "propose", "op": "session.propose", + "unit_id": "unit-demo-l01", + "stage_id": "stage-demo"}]}) + argv = task_ir.compile_spec(propose).plan["steps"][0]["argv"] + assert argv[:3] == ["runtime-session", "--requirement", "req-demo-l01-demo"] + + +def test_workflow_preview_apply_replay_and_isolation(mini_repo, tmp_path): + repo, rid, smid = material_fixture(mini_repo) + before_map = copy.deepcopy(repo.module_source_maps["module-demo"]) + before_stage = copy.deepcopy(repo.study_maps[smid].data["stages"][0]) + spec = task_ir.parse_spec(_spec()) + evidence = tmp_path / "evidence" + fingerprint = canonical_fingerprint(mini_repo) + + preview = task_ir.preview_task(mini_repo, spec, evidence, run_reads=_run_reads) + assert preview["status"] == "needs_review" + by_id = {row["id"]: row for row in preview["operations"]} + assert by_id["find-material"]["outputs"]["route_id"] == rid + assert by_id["patch-route"]["status"] == "needs_review" + # Preview purity: zero canonical writes, zero receipts. + assert canonical_fingerprint(mini_repo) == fingerprint + assert _receipts(mini_repo) == [] + + applied = task_ir.apply_task(mini_repo, spec, evidence) + assert applied["status"] == "committed" + assert (mini_repo / applied["receipt"]["receipt_path"]).is_file() + assert {row["id"]: row for row in applied["operations"]}["patch-route"]["status"] == "committed" + + changed = load_repo(mini_repo) + route = changed.module_source_maps["module-demo"]["sources"][0]["unit_routes"][0] + assert route["angle"] == "A task-ir angle" + # Field isolation: every other route field and the stage override survive. + for key, value in before_map["sources"][0]["unit_routes"][0].items(): + if key != "angle": + assert route.get(key) == value + stage = changed.study_maps[smid].data["stages"][0] + assert stage["resources"][1]["angle"] == before_stage["resources"][1]["angle"] + assert stage["resources"][2] == before_stage["resources"][2] + assert len(_receipts(mini_repo)) == 1 + + # Idempotent replay: evidence lookup, no second dispatch, no new receipt. + replay = task_ir.apply_task(mini_repo, spec, evidence) + assert replay["status"] == "committed" and replay["resumed"] is True + assert replay["receipt"]["transaction_id"] == applied["receipt"]["transaction_id"] + assert len(_receipts(mini_repo)) == 1 + + +def test_unchanged_set_fields_never_dispatch(mini_repo, tmp_path): + repo, _, _ = material_fixture(mini_repo) + route = repo.module_source_maps["module-demo"]["sources"][0]["unit_routes"][0] + raw = _spec() + raw["operations"][2]["set_fields"] = {"angle": route["angle"]} + spec = task_ir.parse_spec(raw) + evidence = tmp_path / "evidence" + preview = task_ir.preview_task(mini_repo, spec, evidence, run_reads=_run_reads) + assert preview["status"] == "unchanged" + applied = task_ir.apply_task(mini_repo, spec, evidence) + assert applied["status"] == "unchanged" + assert _receipts(mini_repo) == [] + + +def test_stale_snapshot_refuses_apply(mini_repo, tmp_path): + repo, rid, smid = material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + evidence = tmp_path / "evidence" + assert task_ir.preview_task(mini_repo, spec, evidence, run_reads=_run_reads)["status"] == "needs_review" + # An unrelated reviewed write moves the snapshot after preview. + other = task_ir.parse_spec({**_spec(), "request_id": "task-test-other"}) + other_evidence = tmp_path / "other-evidence" + assert task_ir.preview_task(mini_repo, other, other_evidence, run_reads=_run_reads)["status"] == "needs_review" + assert task_ir.apply_task(mini_repo, other, other_evidence)["status"] == "committed" + refused = task_ir.apply_task(mini_repo, spec, evidence) + assert refused["status"] == "refused" + assert len(_receipts(mini_repo)) == 1 # only the unrelated write committed + + +def test_request_id_reuse_with_different_spec_is_an_error(mini_repo, tmp_path): + material_fixture(mini_repo) + evidence = tmp_path / "evidence" + spec = task_ir.parse_spec(_spec()) + task_ir.preview_task(mini_repo, spec, evidence, run_reads=_run_reads) + changed = _spec() + changed["operations"][2]["set_fields"] = {"angle": "Another angle"} + with pytest.raises(task_ir.TaskSpecError, match="different spec"): + task_ir.apply_task(mini_repo, task_ir.parse_spec(changed), evidence) + + +def test_interrupted_dispatch_is_uncertain_and_never_redispatched(mini_repo, tmp_path): + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + evidence = tmp_path / "evidence" + calls = [] + + def exploding_submit(root, record, payload, key, request_id): + calls.append(key) + raise TimeoutError("lost after dispatch") + + task_ir.preview_task(mini_repo, spec, evidence, run_reads=_run_reads) + first = task_ir.apply_task(mini_repo, spec, evidence, seal_submit=exploding_submit) + assert first["status"] == "uncertain" + second = task_ir.apply_task(mini_repo, spec, evidence, seal_submit=exploding_submit) + assert second["status"] == "uncertain" and second["resumed"] is True + assert len(calls) == 1 + assert _receipts(mini_repo) == [] + + +def test_preview_check_failure_is_refused_before_any_dispatch(mini_repo, tmp_path): + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + evidence = tmp_path / "evidence" + + def failing_check(root, payload, key, record_path): + return 2, "", "seal_envelope: refused by fixture" + + preview = task_ir.preview_task(mini_repo, spec, evidence, run_reads=_run_reads, + seal_check=failing_check) + assert preview["status"] == "partial" + assert {r["id"]: r for r in preview["operations"]}["patch-route"]["status"] == "refused" + assert _receipts(mini_repo) == [] + + +def test_session_propose_without_runtime_target_refuses_cleanly(mini_repo, tmp_path): + material_fixture(mini_repo) + spec = task_ir.parse_spec({"schema_version": 1, "request_id": "task-test-3", + "operations": [{"id": "propose", "op": "session.propose", + "unit_id": "unit-demo-l01", + "stage_id": "stage-demo"}]}) + preview = task_ir.preview_task(mini_repo, spec, tmp_path / "evidence", run_reads=_run_reads) + assert preview["operations"][0]["status"] == "refused" + + +def test_cli_preview_writes_nothing(mini_repo, tmp_path): + material_fixture(mini_repo) + spec_path = tmp_path / "spec.json" + spec_path.write_text(json.dumps(_spec()), encoding="utf-8") + fingerprint = canonical_fingerprint(mini_repo) + proc = subprocess.run( + [sys.executable, str(ROOT / "tools/agent_run.py"), "--task", str(spec_path), + "--root", str(mini_repo), "--evidence-dir", str(tmp_path / "evidence")], + capture_output=True, text=True, timeout=600) + assert proc.returncode == 0, proc.stdout + proc.stderr + report = json.loads(proc.stdout) + assert report["status"] == "needs_review" + assert canonical_fingerprint(mini_repo) == fingerprint + assert _receipts(mini_repo) == [] + verify = run_los(mini_repo, "plan-edit-context", "unit-demo-l01", "--route-id", + report["operations"][1]["outputs"]["route_id"]) + assert verify.returncode == 0, verify.stderr From d707aaa13c63633d7bc5407ebf11875d1a4f2450 Mon Sep 17 00:00:00 2001 From: Aram Aljanadi Date: Tue, 27 Oct 2026 23:08:56 +0100 Subject: [PATCH 3/5] Repair Task IR P1 evidence persistence and transcript verification --- system/procedures/add-or-evaluate-a-source.md | 2 +- tests/test_task_ir.py | 106 ++++++++++++++++++ tools/learning_os/app/task_ir.py | 43 +++++-- ...2026-10-27-task-ir-evidence-persistence.md | 10 ++ .../2026-10-27-task-ir-open-read-shapes.md | 11 ++ .../2026-10-27-task-ir-read-results.md | 11 ++ .../2026-10-27-task-ir-transcript-checksum.md | 10 ++ ...-10-27-task-ir-uncertain-reconciliation.md | 11 ++ 8 files changed, 195 insertions(+), 9 deletions(-) create mode 100644 work/complaints/2026-10-27-task-ir-evidence-persistence.md create mode 100644 work/complaints/2026-10-27-task-ir-open-read-shapes.md create mode 100644 work/complaints/2026-10-27-task-ir-read-results.md create mode 100644 work/complaints/2026-10-27-task-ir-transcript-checksum.md create mode 100644 work/complaints/2026-10-27-task-ir-uncertain-reconciliation.md diff --git a/system/procedures/add-or-evaluate-a-source.md b/system/procedures/add-or-evaluate-a-source.md index f6ff9d3..bd066f6 100644 --- a/system/procedures/add-or-evaluate-a-source.md +++ b/system/procedures/add-or-evaluate-a-source.md @@ -188,7 +188,7 @@ any destination outside the transcript `TMPDIR`. ```scrut $ mkdir -p "$TMPDIR/inv-pair/repo" && export INV_PAIR="$(realpath "$TMPDIR/inv-pair")" INV_REPO="$(realpath "$TMPDIR/inv-pair/repo")" && $H/los-anchored-copy materials_manifest "$INV_REPO" sha256:a0583ab05765120a52f688d88ed6cd03e56c1567d7183328d0c640424297b575 Makefile -sha256:6b5296af66bf7bfb1a03b3dbb6c8ee6778fa4e7ba121fac957d73d8502f14161 pyproject.toml +sha256:3ceabba96c45022e7f726ab432e918da7769554aac8b2f757f0db9033955cef0 pyproject.toml sha256:0c84a5cae52dc8aa786e0009eaaa38e9e2f202b5f140e8cb07d1b188bc338ba8 tools/learning_os/__init__.py sha256:9f4511b38e35ff56efa15bc8b486a589d9dc42eed38ab331da85c7e7f6503c41 tools/learning_os/domain/material_inventory.py sha256:8bdcc529afb1448f8856ba13e615b894989c70786024d702e839f62b52e96e3b tools/materials_manifest.py diff --git a/tests/test_task_ir.py b/tests/test_task_ir.py index 982b702..93c6ac3 100644 --- a/tests/test_task_ir.py +++ b/tests/test_task_ir.py @@ -3,6 +3,7 @@ from __future__ import annotations import copy +import hashlib import json import subprocess import sys @@ -221,3 +222,108 @@ def test_cli_preview_writes_nothing(mini_repo, tmp_path): verify = run_los(mini_repo, "plan-edit-context", "unit-demo-l01", "--route-id", report["operations"][1]["outputs"]["route_id"]) assert verify.returncode == 0, verify.stderr + + +@pytest.mark.parametrize("phase", ["preview", "apply"]) +@pytest.mark.parametrize("destination", ["evidence", "temporary", "check"]) +def test_evidence_symlinks_never_touch_canonical_files(mini_repo, tmp_path, phase, destination): + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + directory = tmp_path / "evidence" + directory.mkdir() + if phase == "apply": + assert task_ir.preview_task(mini_repo, spec, directory, + run_reads=_run_reads)["status"] == "needs_review" + # The request evidence uses a hashed name, separate from its check record. + evidence_path = directory / f"task-{hashlib.sha256(spec.request_id.encode()).hexdigest()[:32]}.json" + paths = {"evidence": evidence_path, + "temporary": evidence_path.with_name(f".{evidence_path.name}.tmp"), + "check": directory / f"task-{spec.request_id}-patch-route.check.json"} + redirected = paths[destination] + redirected.unlink(missing_ok=True) + redirected.symlink_to(mini_repo / "sources/sources.yaml") + before = canonical_fingerprint(mini_repo) + calls = [] + + def submit(*args): + calls.append(args) + pytest.fail("redirected evidence must be refused before submission") + + with pytest.raises(task_ir.TaskSpecError, match="symlink") as exc: + if phase == "preview": + task_ir.preview_task(mini_repo, spec, directory, run_reads=_run_reads) + else: + task_ir.apply_task(mini_repo, spec, directory, seal_submit=submit) + assert exc.value.code == "TASK_IR_EVIDENCE" + assert calls == [] + assert canonical_fingerprint(mini_repo) == before + assert _receipts(mini_repo) == [] + + +def test_evidence_directory_cannot_resolve_into_data_root(mini_repo, tmp_path): + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + directory = tmp_path / "redirected-evidence" + directory.symlink_to(mini_repo, target_is_directory=True) + before = canonical_fingerprint(mini_repo) + for phase in (task_ir.preview_task, task_ir.apply_task): + kwargs = {"run_reads": _run_reads} if phase is task_ir.preview_task else {} + with pytest.raises(task_ir.TaskSpecError, match="outside the data root"): + phase(mini_repo, spec, directory, **kwargs) + assert canonical_fingerprint(mini_repo) == before + assert _receipts(mini_repo) == [] + + +def test_failed_checkpoint_publication_prevents_dispatch_and_keeps_preview(mini_repo, tmp_path, monkeypatch): + from learning_os.store import transactions + + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + directory = tmp_path / "evidence" + assert task_ir.preview_task(mini_repo, spec, directory, + run_reads=_run_reads)["status"] == "needs_review" + evidence_path = next(path for path in directory.glob("task-*.json") + if not path.name.endswith(".check.json")) + before_bytes = evidence_path.read_bytes() + before = canonical_fingerprint(mini_repo) + calls = [] + replace = transactions.os.replace + + def fail_publication(source, target): + if target == evidence_path: + raise OSError("simulated interruption before checkpoint publication") + return replace(source, target) + + def submit(*args): + calls.append(args) + pytest.fail("dispatch must wait for its durable checkpoint") + + with monkeypatch.context() as patch: + patch.setattr(transactions.os, "replace", fail_publication) + with pytest.raises(task_ir.TaskSpecError, match="could not be persisted") as exc: + task_ir.apply_task(mini_repo, spec, directory, seal_submit=submit) + assert exc.value.code == "TASK_IR_EVIDENCE" + assert calls == [] + assert evidence_path.read_bytes() == before_bytes + assert canonical_fingerprint(mini_repo) == before + assert _receipts(mini_repo) == [] + assert task_ir.apply_task(mini_repo, spec, directory)["status"] == "committed" + assert len(_receipts(mini_repo)) == 1 + + +def test_earlier_relative_check_record_still_applies(mini_repo, tmp_path, monkeypatch): + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + monkeypatch.chdir(tmp_path) + (tmp_path / "nested").mkdir() + directory = Path("nested/../evidence") + assert task_ir.preview_task(mini_repo, spec, directory, + run_reads=_run_reads)["status"] == "needs_review" + evidence_path = next(path for path in directory.glob("task-*.json") + if not path.name.endswith(".check.json")) + evidence = json.loads(evidence_path.read_text()) + # The preceding implementation persisted the caller's relative spelling. + evidence["check_record"] = str(directory / Path(evidence["check_record"]).name) + evidence_path.write_text(json.dumps(evidence)) + assert task_ir.apply_task(mini_repo, spec, directory)["status"] == "committed" + assert len(_receipts(mini_repo)) == 1 diff --git a/tools/learning_os/app/task_ir.py b/tools/learning_os/app/task_ir.py index abfc38e..a3c95a7 100644 --- a/tools/learning_os/app/task_ir.py +++ b/tools/learning_os/app/task_ir.py @@ -54,7 +54,9 @@ from learning_os.app.agent_reporting import encode from learning_os.app.registry import REGISTRY, OperationEffects from learning_os.derive.learning_runtime import requirement_id_for +from learning_os.domain.failures import TransactionFailure from learning_os.domain.material_refs import PATCH_FIELDS +from learning_os.store.transactions import atomic_write_bytes TOOLS = Path(__file__).resolve().parents[2] @@ -383,14 +385,35 @@ def _evidence_path(evidence_dir: Path, request_id: str) -> Path: return evidence_dir / f"task-{name}.json" +def _evidence_directory(root: Path, evidence_dir: Path) -> Path: + directory = evidence_dir.resolve() + if directory.is_relative_to(root.resolve()): + raise _fail("evidence directory must live outside the data root", "TASK_IR_EVIDENCE") + return directory + + +def _check_evidence_path(path: Path) -> None: + """Refuse redirected evidence, including the atomic writer's sibling.""" + if any(part.is_symlink() for part in (path, *path.parents)): + raise _fail(f"evidence path traverses a symlink: {path}", "TASK_IR_EVIDENCE") + if path.exists() and not path.is_file(): + raise _fail(f"evidence path is not a regular file: {path}", "TASK_IR_EVIDENCE") + + def _save_evidence(path: Path, evidence: dict[str, object]) -> None: - path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) - tmp = path.with_suffix(".tmp") - tmp.write_text(encode(evidence) + "\n", encoding="utf-8") - tmp.replace(path) + # As in agent_reporting, this is private recovery evidence, not a + # canonical transaction or disposable cache. Reuse the shared persistence + # barrier so dispatch cannot precede a durable checkpoint. + _check_evidence_path(path) + _check_evidence_path(path.with_name(f".{path.name}.tmp")) + try: + atomic_write_bytes(path, (encode(evidence) + "\n").encode("utf-8")) + except TransactionFailure as exc: + raise _fail(f"run evidence could not be persisted: {exc}", "TASK_IR_EVIDENCE") from exc def _load_evidence(path: Path, spec: TaskSpec) -> dict[str, object] | None: + _check_evidence_path(path) if not path.is_file(): return None try: @@ -461,8 +484,7 @@ def preview_task(root: Path, spec: TaskSpec, evidence_dir: Path, *, run_reads: ReadRunner, seal_check: SealCheck = default_seal_check) -> dict[str, object]: """Run reads and the reviewed check; zero canonical writes, zero receipts.""" - if evidence_dir.resolve().is_relative_to(root.resolve()): - raise _fail("evidence directory must live outside the data root", "TASK_IR_EVIDENCE") + evidence_dir = _evidence_directory(root, evidence_dir) evidence_dir.mkdir(parents=True, exist_ok=True, mode=0o700) path = _evidence_path(evidence_dir, spec.request_id) prior = _load_evidence(path, spec) @@ -503,6 +525,7 @@ def preview_task(root: Path, spec: TaskSpec, evidence_dir: Path, *, payload = write_payload(write, route_id) key = operation_key(spec.request_id, write.id) record_path = evidence_dir / f"task-{spec.request_id}-{write.id}.check.json" + _check_evidence_path(record_path) code, _out, err = seal_check(root, payload, key, record_path) if code != 0 and "no material changes to apply" in err: # The existing machinery itself refuses a no-change patch @@ -545,8 +568,7 @@ def _committed_receipt(root: Path, key: str) -> dict[str, object] | None: def apply_task(root: Path, spec: TaskSpec, evidence_dir: Path, *, seal_submit: SealSubmit = default_seal_submit) -> dict[str, object]: """Apply the previewed write once; retries look up, never redispatch blind.""" - if evidence_dir.resolve().is_relative_to(root.resolve()): - raise _fail("evidence directory must live outside the data root", "TASK_IR_EVIDENCE") + evidence_dir = _evidence_directory(root, evidence_dir) path = _evidence_path(evidence_dir, spec.request_id) evidence = _load_evidence(path, spec) if evidence is None: @@ -559,6 +581,11 @@ def apply_task(root: Path, spec: TaskSpec, evidence_dir: Path, *, assert compiled.write is not None # needs_review is only set with a write key = str(evidence["key"]) record_path = Path(str(evidence["check_record"])) + expected_record = evidence_dir / f"task-{spec.request_id}-{compiled.write.id}.check.json" + _check_evidence_path(expected_record) + if record_path.resolve() != expected_record: + raise _fail("check record must stay in the evidence directory", "TASK_IR_EVIDENCE") + record_path = expected_record payload = evidence["payload"] if not isinstance(payload, dict) or not record_path.is_file(): raise _fail("preview evidence is incomplete", "TASK_IR_EVIDENCE") diff --git a/work/complaints/2026-10-27-task-ir-evidence-persistence.md b/work/complaints/2026-10-27-task-ir-evidence-persistence.md new file mode 100644 index 0000000..7cc4b9e --- /dev/null +++ b/work/complaints/2026-10-27-task-ir-evidence-persistence.md @@ -0,0 +1,10 @@ +# Task IR P1 evidence persistence bypassed the shared atomic writer + +- Workflow: continuing the P1 experiment and running its Core verification. +- Tested state: Core `3541789` on `codex/task-ir-p1-experiment`, paired UI `3c9a5f664c147790c490fae0a256e7764b4b9157`; macOS, Python 3.14.4. Only synthetic fixtures were mutated. +- Expected: private run evidence outside the data root uses the existing atomic persistence primitive, without canonical transaction receipts or a new lint exemption. +- Observed: `tools/learning_os/app/task_ir.py::_save_evidence` called `write_text` and `replace`; `tests/test_arch_writes.py::test_static_lint_matches_baseline` rejected both. The old predictable temporary path could also follow a symlink. The baseline combined command reported 2 failed, 44 passed, 2 xfailed; the other failure was the separately reported transcript hash. +- Smallest repair: reuse `store.transactions.atomic_write_bytes`, already used by `app.agent_reporting`, and reject redirected evidence/check destinations before use. The repair adds no ratchet or effect-classification exemption. +- Verification: `tests/test_task_ir.py` plus `tests/test_arch_writes.py` passed 39 cases, including six preview/apply symlink cases, a directory redirected into the data root, a failed checkpoint publication that prevented submission and retained the previous preview, and compatibility with previously persisted relative check-record paths. The existing file/directory synchronization test also passed. This does not establish protection against concurrent hostile directory replacement or complete P1 recovery acceptance. +- Evidence: `../repair-baseline.log`, `../repair-focused-final.log`, `../repair-durability.log` in the experiment workbench; reproduction commands are retained in `../REPAIR-PLAN.md`. +- GitHub: not filed. Automatic approval review rejected publishing the first prepared issue to `Aram164/LearningOS`, stating that the specific contents and destination require explicit approval; the remaining issue submissions were halted. The exact ready-to-file body is this note, without this final status line. diff --git a/work/complaints/2026-10-27-task-ir-open-read-shapes.md b/work/complaints/2026-10-27-task-ir-open-read-shapes.md new file mode 100644 index 0000000..f2cb613 --- /dev/null +++ b/work/complaints/2026-10-27-task-ir-open-read-shapes.md @@ -0,0 +1,11 @@ +# Task IR read operations silently accept undeclared fields + +- Workflow: requested architectural assessment of Task IR P1 before acceptance. +- Tested state: Core `3541789` plus the isolated evidence/checksum repair; `_operation` read branches are unchanged. Python 3.14.4, no learner writes. +- Expected: every operation has a closed input contract and refuses undeclared fields before execution. +- Observed: `_operation` in `tools/learning_os/app/task_ir.py` returns from all three read branches before its final extra-field check. `parse_spec` accepts `surprise: true` on `context.inspect`, `material.find`, and `session.propose`. It retains the field in the normalized digest while execution ignores it. +- Reproduction: call `parse_spec` with schema version 1, a request ID, and one valid read operation carrying `surprise: true`. Bounded output in workbench `assessment-parser.log` records all three acceptances. +- Impact: a misspelled constraint can be silently ignored; this is a compile-contract failure, not evidence of a gateway authorization bypass. +- Smallest improvement: apply the existing per-operation allowed-field check before every return. Add real-interface refusal cases for each read operation without creating another capability catalogue. +- Acceptance: all three read operations reject unknown fields before dispatch, while valid existing specifications still compile deterministically. No repair for this gap is included in the immediate verification increment. +- GitHub: not filed. Automatic approval review rejected publishing the first prepared issue to `Aram164/LearningOS`, stating that the specific contents and destination require explicit approval; the remaining issue submissions were halted. The exact ready-to-file body is this note, without this final status line. diff --git a/work/complaints/2026-10-27-task-ir-read-results.md b/work/complaints/2026-10-27-task-ir-read-results.md new file mode 100644 index 0000000..34608ee --- /dev/null +++ b/work/complaints/2026-10-27-task-ir-read-results.md @@ -0,0 +1,11 @@ +# Successful read-only Task IR tasks return an unrecognized status and lose useful context + +- Workflow: requested P1 architecture assessment through `tools/agent_run.py --task` against a synthetic fixture. +- Tested state: Core `3541789` plus the isolated evidence/checksum repair; read-result projection and CLI exit mapping are unchanged. macOS/Python 3.14.4. +- Expected: successful read-only tasks exit successfully and return bounded useful context with its snapshot. +- Observed: a task containing valid `context.inspect` and `material.find` reports both operation statuses `ok`, top-level status `previewed`, and CLI exit 3. Canonical fingerprint is unchanged. The inspect output contains only the record ID and a null snapshot. `_outputs_from_report` also drops the runtime contract from `session.propose`; that loss is verified by code inspection, not a positive authored-runtime integration run. +- Reproduction: run workbench `assessment_probe.py`; `assessment-integration.log` retains the actual CLI report and exit code. +- Impact: callers treat a successful read-only task as failure, and inspecting context or proposing a session cannot deliver its central result through the public task report. +- Smallest improvement: settle read-only tasks from their actual operation results and align the CLI mapping; carry bounded inspect/session projections and snapshots from the existing runner. Reuse existing full-report references for larger content. +- Acceptance: real CLI tests distinguish successful and refused read-only tasks; a valid authored runtime target produces a snapshot-bound session contract without evidence/mastery claims or canonical changes. No new session entity or requirement store. This remains later P1 acceptance work. +- GitHub: not filed. Automatic approval review rejected publishing the first prepared issue to `Aram164/LearningOS`, stating that the specific contents and destination require explicit approval; the remaining issue submissions were halted. The exact ready-to-file body is this note, without this final status line. diff --git a/work/complaints/2026-10-27-task-ir-transcript-checksum.md b/work/complaints/2026-10-27-task-ir-transcript-checksum.md new file mode 100644 index 0000000..65e9265 --- /dev/null +++ b/work/complaints/2026-10-27-task-ir-transcript-checksum.md @@ -0,0 +1,10 @@ +# P1 strict type-checking addition invalidated an anchored transcript checksum + +- Workflow: Core procedure verification after the Task IR P1 implementation. +- Tested state: Core `3541789`; isolated experiment, macOS/Python 3.14.4. +- Expected: the anchored-copy transcript records the exact current `pyproject.toml` bytes while preserving byte-for-byte fixture copying and checksum checking. +- Observed: `ea00749` added `tools/learning_os/app/task_ir.py` to the strict mypy file list. `system/procedures/add-or-evaluate-a-source.md:191` retained the prior hash, so `tests/test_procedures.py::test_every_transcript_passes` failed. Measured current hash: `sha256:3ceabba96c45022e7f726ab432e918da7769554aac8b2f757f0db9033955cef0`. +- Smallest repair: retain strict mypy coverage and update only the transcript's expected pyproject hash. Do not weaken `tests/procedure_helpers/los-anchored-copy` or add wildcard output. +- Acceptance: the exact hash matches `shasum -a 256 pyproject.toml`, and the complete procedure suite passes with its existing audit witnesses and expected failures. Final counts are retained in the workbench report. +- Evidence: `../repair-baseline.log`, `../repair-procedures.log`, and `../REPAIR-REPORT.md` in the experiment workbench. +- GitHub: not filed. Automatic approval review rejected publishing the first prepared issue to `Aram164/LearningOS`, stating that the specific contents and destination require explicit approval; the remaining issue submissions were halted. The exact ready-to-file body is this note, without this final status line. diff --git a/work/complaints/2026-10-27-task-ir-uncertain-reconciliation.md b/work/complaints/2026-10-27-task-ir-uncertain-reconciliation.md new file mode 100644 index 0000000..c0c3f01 --- /dev/null +++ b/work/complaints/2026-10-27-task-ir-uncertain-reconciliation.md @@ -0,0 +1,11 @@ +# Task IR cannot reconcile an already committed write after its response is lost + +- Workflow: requested P1 recovery-boundary assessment using a disposable synthetic material fixture. +- Tested state: Core `3541789` plus the isolated evidence/checksum repair; the apply recovery branches are unchanged. macOS/Python 3.14.4; no actual learner data changed. +- Expected: retain the original request, idempotency key and sealed envelope; reconcile the existing transaction and report its receipt without a fresh write. +- Observed: wrap `default_seal_submit` so it completes a real synthetic route patch and then raises `TimeoutError`. `apply_task` reports `uncertain`; calling it again returns `uncertain` without receipt reconciliation. There is exactly one receipt and one adapter dispatch. Calling the underlying seal submission with the original check, payload, key and request ID reports `replayed: true`, the same transaction ID, and still one receipt. +- Proof: `../assessment_probe.py` and `../assessment-integration.log` in the workbench. Output: first/second status `uncertain`, dispatch count 1, receipt count 1, underlying exact replay true, same transaction true. Code inspection also shows persisted `dispatched` is returned without reconciliation after an abrupt process exit; that abrupt-exit case has not been executed here. +- Impact: the adapter safely avoids blind redispatch but strands recoverable committed outcomes. Its current test proves non-redispatch, not completed recovery. +- Smallest improvement: use the existing receipt/retained-envelope recovery mechanism for uncertain/dispatched states, preserve exact identity, and define explicit operator action when evidence cannot establish an outcome. No second journal or resealing protocol. +- Acceptance: adapter-level lost-response, abrupt-exit, missing-receipt, conflicting-intent and exact-replay cases prove one transaction, one receipt, preserved identities, and truthful unresolved outcomes. This is deferred acceptance work. +- GitHub: not filed. Automatic approval review rejected publishing the first prepared issue to `Aram164/LearningOS`, stating that the specific contents and destination require explicit approval; the remaining issue submissions were halted. The exact ready-to-file body is this note, without this final status line. From 7b3481012747730584d49c27c395d226807f634c Mon Sep 17 00:00:00 2001 From: Aram Aljanadi Date: Tue, 27 Oct 2026 23:39:49 +0100 Subject: [PATCH 4/5] Complete Task IR P1 acceptance and exact-envelope recovery Close read shapes and retain bounded context, material provenance and runtime session proposals under verified read snapshots. Bind reviewed artifacts to the task and submit the original envelope bytes; reconcile interrupted writes through replay-only and strict receipt verification. Cover authored-runtime, scope/guard/review substitution, interruption, replay and persistence cases. LOS-CI-Bootstrap: codex/task-ir-p1-experiment --- .../components/operation-pipeline.md | 11 + tests/test_task_ir.py | 488 +++++++++++++++++- tools/agent_run.py | 2 +- tools/learning_os/app/task_ir.py | 353 +++++++++---- ...2026-10-27-task-ir-evidence-persistence.md | 2 +- .../2026-10-27-task-ir-false-unchanged.md | 9 + .../2026-10-27-task-ir-hardlink-preview.md | 9 + .../2026-10-27-task-ir-open-read-shapes.md | 4 +- .../2026-10-27-task-ir-read-results.md | 4 +- .../2026-10-27-task-ir-transcript-checksum.md | 2 +- ...-10-27-task-ir-uncertain-reconciliation.md | 4 +- 11 files changed, 762 insertions(+), 126 deletions(-) create mode 100644 work/complaints/2026-10-27-task-ir-false-unchanged.md create mode 100644 work/complaints/2026-10-27-task-ir-hardlink-preview.md diff --git a/system/developer/components/operation-pipeline.md b/system/developer/components/operation-pipeline.md index 97c0697..697f94a 100644 --- a/system/developer/components/operation-pipeline.md +++ b/system/developer/components/operation-pipeline.md @@ -59,6 +59,17 @@ Admission lives in `system/contracts/capabilities.yaml`; `app/admission.py` cons `GESTURE_ALLOWLIST` and `UI_REVIEWED_ALLOWLIST` enforce it. The UI mirrors its permitted gesture set and checks parity in its own contract check. +### Experimental Task IR P1 + +`task_ir.py` compiles four closed operations into the existing runner/gateway. +Read results retain bounded content, snapshot pins and expansion references; route binding requires complete discovery. +Private evidence retains original check/envelope bytes with digests and root/session/request bindings through the shared atomic writer. +Apply submits copied original bytes; interrupted outcomes use gateway replay-only and the strict committed-evidence verifier. Missing proof stays uncertain; +preview never replaces in-flight evidence, and interrupted requests are never +resealed. Legacy previews lacking original bytes require a new reviewed request ID. +Operator assertions do not authenticate human review. There is no new journal, session entity, approval protocol or general graph composition. +Pinning: `tests/test_task_ir.py`. + ## Pinning tests - `tests/test_ops_pipeline.py` diff --git a/tests/test_task_ir.py b/tests/test_task_ir.py index 93c6ac3..adb174b 100644 --- a/tests/test_task_ir.py +++ b/tests/test_task_ir.py @@ -5,17 +5,21 @@ import copy import hashlib import json +import os import subprocess import sys from pathlib import Path import pytest -from repo_builders import material_fixture, run_los +import yaml +from repo_builders import material_fixture, run_los, write_yaml from learning_os.adapters import agent_runner from learning_os.app import task_ir from learning_os.app.agent_reporting import encode +from learning_os.derive.learning_runtime import collect_requirements, requirement_id_for from learning_os.store.fingerprint import canonical_fingerprint +from learning_os.store.fingerprint import snapshot as current_snapshot from learning_os.store.loader import load_repo ROOT = Path(__file__).resolve().parents[1] @@ -112,6 +116,12 @@ def test_workflow_preview_apply_replay_and_isolation(mini_repo, tmp_path): stage = changed.study_maps[smid].data["stages"][0] assert stage["resources"][1]["angle"] == before_stage["resources"][1]["angle"] assert stage["resources"][2] == before_stage["resources"][2] + expected_map = copy.deepcopy(before_map) + expected_map["sources"][0]["unit_routes"][0].update(angle="A task-ir angle", id=rid) + assert changed.module_source_maps["module-demo"] == expected_map + assert {k: v for k, v in stage.items() if k != "resources"} == { + k: v for k, v in before_stage.items() if k != "resources"} + assert applied["operations"][-1]["outputs"]["transaction_id"] == applied["receipt"]["transaction_id"] assert len(_receipts(mini_repo)) == 1 # Idempotent replay: evidence lookup, no second dispatch, no new receipt. @@ -167,8 +177,8 @@ def test_interrupted_dispatch_is_uncertain_and_never_redispatched(mini_repo, tmp evidence = tmp_path / "evidence" calls = [] - def exploding_submit(root, record, payload, key, request_id): - calls.append(key) + def exploding_submit(root, content): + calls.append(json.loads(content)["idempotency_key"]) raise TimeoutError("lost after dispatch") task_ir.preview_task(mini_repo, spec, evidence, run_reads=_run_reads) @@ -185,7 +195,7 @@ def test_preview_check_failure_is_refused_before_any_dispatch(mini_repo, tmp_pat spec = task_ir.parse_spec(_spec()) evidence = tmp_path / "evidence" - def failing_check(root, payload, key, record_path): + def failing_check(root, payload, key, record_path, session_id): return 2, "", "seal_envelope: refused by fixture" preview = task_ir.preview_task(mini_repo, spec, evidence, run_reads=_run_reads, @@ -283,7 +293,7 @@ def test_failed_checkpoint_publication_prevents_dispatch_and_keeps_preview(mini_ assert task_ir.preview_task(mini_repo, spec, directory, run_reads=_run_reads)["status"] == "needs_review" evidence_path = next(path for path in directory.glob("task-*.json") - if not path.name.endswith(".check.json")) + if not path.name.endswith((".check.json", ".envelope.json"))) before_bytes = evidence_path.read_bytes() before = canonical_fingerprint(mini_repo) calls = [] @@ -320,10 +330,476 @@ def test_earlier_relative_check_record_still_applies(mini_repo, tmp_path, monkey assert task_ir.preview_task(mini_repo, spec, directory, run_reads=_run_reads)["status"] == "needs_review" evidence_path = next(path for path in directory.glob("task-*.json") - if not path.name.endswith(".check.json")) + if not path.name.endswith((".check.json", ".envelope.json"))) evidence = json.loads(evidence_path.read_text()) # The preceding implementation persisted the caller's relative spelling. evidence["check_record"] = str(directory / Path(evidence["check_record"]).name) evidence_path.write_text(json.dumps(evidence)) assert task_ir.apply_task(mini_repo, spec, directory)["status"] == "committed" assert len(_receipts(mini_repo)) == 1 + + +def _state(directory): + path = next(path for path in directory.glob("task-*.json") + if not path.name.endswith((".check.json", ".envelope.json"))) + return path, json.loads(path.read_text()) + + +@pytest.mark.parametrize("op,fields", [ + ("context.inspect", {"unit_id": "unit-demo-l01"}), + ("material.find", {"unit_id": "unit-demo-l01"}), + ("session.propose", {"unit_id": "unit-demo-l01", "stage_id": "stage-demo"}), +]) +def test_every_read_operation_refuses_undeclared_fields(op, fields): + with pytest.raises(task_ir.TaskSpecError, match="undeclared"): + task_ir.parse_spec({"schema_version": 1, "request_id": "closed-read", "operations": [ + {"id": "read", "op": op, **fields, "surprise": True}]}) + + +@pytest.mark.parametrize("invalid", [True, "1", 1.0]) +def test_schema_version_is_an_integer_contract(invalid): + with pytest.raises(task_ir.TaskSpecError): + task_ir.parse_spec(_spec(schema_version=invalid)) + + +def test_cli_successful_reads_expose_bounded_context_and_snapshot(mini_repo, tmp_path): + material_fixture(mini_repo) + raw = _spec() + raw["operations"].pop() + source = tmp_path / "read.json" + source.write_text(json.dumps(raw)) + before = canonical_fingerprint(mini_repo) + proc = subprocess.run([sys.executable, str(ROOT / "tools/agent_run.py"), "--task", str(source), + "--root", str(mini_repo), "--evidence-dir", str(tmp_path / "evidence")], + capture_output=True, text=True, timeout=60) + assert proc.returncode == 0, proc.stdout + proc.stderr + result = json.loads(proc.stdout) + assert result["status"] == "ok" + inspect, materials = [row["outputs"] for row in result["operations"]] + assert inspect["context"]["title"] == "Expected value" + assert inspect["snapshot_id"] == materials["snapshot_id"] == current_snapshot(mini_repo) + assert materials["complete"] is True + resources = materials["materials"]["stage"]["resources"] + assert len(resources) == 3 # Distinct authored placements survive route-id deduplication. + assert resources[0]["source_id"] == "source-demo-book" + assert resources[0]["angle"] != resources[1]["angle"] + assert canonical_fingerprint(mini_repo) == before + assert not _receipts(mini_repo) + + +@pytest.mark.parametrize("target", ["valid", "missing", "invalid"]) +def test_session_proposal_uses_authored_runtime_and_preserves_learner_authority(mini_repo, tmp_path, target): + repo, _, smid = material_fixture(mini_repo) + path = repo.study_maps[smid].path + data = yaml.safe_load(path.read_text()) + authored = {"concept": "concept-expected-value", + "capability": {"kind": "explain", "operands": ["expectation"]}, + "conditions": ["unfamiliar-example"], "evidence_spec": ["explain-reason"]} + if target != "missing": + data["stages"][0]["runtime_target"] = {**authored, **({"evidence_spec": []} if target == "invalid" else {})} + write_yaml(path, data) + spec = task_ir.parse_spec({"schema_version": 1, "request_id": "runtime-proposal", "operations": [ + {"id": "propose", "op": "session.propose", "unit_id": "unit-demo-l01", "stage_id": "stage-demo"}]}) + before = canonical_fingerprint(mini_repo) + report = task_ir.preview_task(mini_repo, spec, tmp_path / "evidence", run_reads=_run_reads) + if target == "valid": + requirement, = collect_requirements(load_repo(mini_repo)) + expected_id = requirement_id_for("unit-demo-l01", "stage-demo") + assert requirement["id"] == expected_id == "req-demo-l01-demo" + assert requirement["source_stage"] == {"module_id": "module-demo", "unit_id": "unit-demo-l01", "stage_id": "stage-demo"} + assert report["status"] == "ok" + outputs = report["operations"][0]["outputs"] + assert outputs["requirement_id"] == expected_id + assert outputs["snapshot_id"] == current_snapshot(mini_repo) + proposal = outputs["proposal"] + direct = run_los(mini_repo, "runtime-session", "--requirement", expected_id) + assert direct.returncode == 0, direct.stderr + assert proposal == json.loads(direct.stdout) + assert proposal["contract"] == "runtime-session-v1" + assert proposal["session"]["source_stage"] == requirement["source_stage"] + assert proposal["session"]["current_status"] == "unseen" + assert proposal["session"]["evidence_ids"] == [] + assert proposal["session"]["target_requirement_id"] == expected_id + assert collect_requirements(load_repo(mini_repo)) == [requirement] + else: + assert report["status"] == "refused" + assert report["operations"][0]["status"] == "refused" + assert canonical_fingerprint(mini_repo) == before + assert not _receipts(mini_repo) + + +@pytest.mark.parametrize("artifact,mutation", [ + ("check", "missing"), ("check", "invalid"), ("check", "payload"), + ("envelope", "missing"), ("envelope", "payload"), ("envelope", "route"), + ("envelope", "unit"), ("envelope", "approval"), + ("state", "payload"), ("state", "route"), ("state", "unit"), ("state", "key"), +]) +def test_review_artifact_substitution_refuses_before_dispatch(mini_repo, tmp_path, artifact, mutation): + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + directory = tmp_path / "evidence" + assert task_ir.preview_task(mini_repo, spec, directory, run_reads=_run_reads)["status"] == "needs_review" + state_path, state = _state(directory) + selected = state_path if artifact == "state" else directory / f"task-{spec.request_id}-patch-route.{artifact}.json" + value = json.loads(selected.read_text()) + if mutation == "missing": + selected.unlink() + elif mutation == "invalid": + selected.write_text("not-json") + else: + payload = value if artifact == "state" else value.get("payload", value) + if mutation == "payload": + if artifact == "state": + value["payload"]["changes"]["angle"] = "substituted" + else: + payload["changes"]["angle"] = "substituted" + elif mutation == "route": + payload["route_id"] = "route-another" + elif mutation == "unit": + if artifact == "state": + value["payload"]["unit_id"] = "unit-another" + else: + payload["unit_id"] = "unit-another" + elif mutation == "key": + value["key"] = "another-key" + else: + value["approval"]["kind"] = "direct-user-gesture" + selected.write_text(json.dumps(value)) + before = canonical_fingerprint(mini_repo) + + def forbidden(*args): + pytest.fail("unreviewed substitution reached dispatch") + + with pytest.raises(task_ir.TaskSpecError): + task_ir.apply_task(mini_repo, spec, directory, seal_submit=forbidden) + assert canonical_fingerprint(mini_repo) == before + assert not _receipts(mini_repo) + + +def test_stale_artifact_revision_refuses_without_a_snapshot_change(mini_repo, tmp_path): + from learning_os.store.revisions import dump_revisions + + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + directory = tmp_path / "evidence" + task_ir.preview_task(mini_repo, spec, directory, run_reads=_run_reads) + _, state = _state(directory) + envelope = json.loads(next(directory.glob("*.envelope.json")).read_text()) + artifact, revision = next(iter(envelope["expected_revisions"].items())) + before = canonical_fingerprint(mini_repo) + target = mini_repo / "operations/transactions/revisions.yaml" + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(dump_revisions({artifact: revision + 1})) + assert canonical_fingerprint(mini_repo) == before + responses = [] + + def submit(root, content): + result = task_ir.default_seal_submit(root, content) + responses.append(json.loads(result[1])) + return result + + assert task_ir.apply_task(mini_repo, spec, directory, seal_submit=submit)["status"] == "refused" + assert responses[0]["error"]["code"] == "REVISION_CONFLICT" + assert not _receipts(mini_repo) + assert canonical_fingerprint(mini_repo) == before + assert state["payload"]["changes"] == {"angle": "A task-ir angle"} + + +def test_lost_committed_response_reconciles_the_original_envelope(mini_repo, tmp_path): + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + directory = tmp_path / "evidence" + preview = task_ir.preview_task(mini_repo, spec, directory, run_reads=_run_reads) + assert preview["status"] == "needs_review" + original = next(directory.glob("*.envelope.json")).read_bytes() + submitted, replayed = [], [] + + def lost_response(root, content): + submitted.append(content) + code, out, err = task_ir.default_seal_submit(root, content) + assert code == 0, err + raise TimeoutError("reply lost after the real commit") + + assert task_ir.apply_task(mini_repo, spec, directory, seal_submit=lost_response)["status"] == "uncertain" + after_commit = canonical_fingerprint(mini_repo) + _, before = _state(directory) + assert task_ir.preview_task(mini_repo, spec, directory, run_reads=_run_reads)["status"] == "uncertain" + assert _state(directory)[1] == before + + def replay(root, content): + replayed.append(content) + return task_ir.default_seal_replay(root, content) + + def forbidden(*args): + pytest.fail("uncertain write was blindly resubmitted") + + recovered = task_ir.apply_task(mini_repo, spec, directory, seal_submit=forbidden, seal_replay=replay) + repeated = task_ir.apply_task(mini_repo, spec, directory, seal_submit=forbidden, seal_replay=replay) + assert recovered["status"] == repeated["status"] == "committed" + assert recovered["receipt"] == repeated["receipt"] + assert recovered["receipt"]["replayed"] is True + assert submitted == [original] + assert replayed == [original, original] + assert next(directory.glob("*.envelope.json")).read_bytes() == original + assert len(_receipts(mini_repo)) == 1 + assert canonical_fingerprint(mini_repo) == after_commit + + +@pytest.mark.parametrize("commit_before_exit", [False, True]) +def test_abrupt_process_exit_preserves_dispatched_evidence(mini_repo, tmp_path, commit_before_exit): + material_fixture(mini_repo) + raw = _spec() + spec = task_ir.parse_spec(raw) + directory = tmp_path / "evidence" + task_ir.preview_task(mini_repo, spec, directory, run_reads=_run_reads) + source = tmp_path / "spec.json" + source.write_text(json.dumps(raw)) + program = ''' +import json, os, sys +from pathlib import Path +sys.path.insert(0, sys.argv[1]) +from learning_os.app import task_ir +def interrupted(root, content): + if sys.argv[5] == "yes": + code, out, err = task_ir.default_seal_submit(root, content) + assert code == 0, err + os._exit(19) +spec = task_ir.parse_spec(json.loads(Path(sys.argv[3]).read_text())) +task_ir.apply_task(Path(sys.argv[2]), spec, Path(sys.argv[4]), seal_submit=interrupted) +''' + proc = subprocess.run([sys.executable, "-c", program, str(ROOT / "tools"), str(mini_repo), + str(source), str(directory), "yes" if commit_before_exit else "no"], + capture_output=True, text=True, timeout=60) + assert proc.returncode == 19, proc.stdout + proc.stderr + state_path, state = _state(directory) + assert state["status"] == "dispatched" + before = state_path.read_bytes() + assert task_ir.preview_task(mini_repo, spec, directory, run_reads=_run_reads)["status"] == "dispatched" + assert state_path.read_bytes() == before + result = task_ir.apply_task(mini_repo, spec, directory) + assert result["status"] == ("committed" if commit_before_exit else "uncertain") + assert len(_receipts(mini_repo)) == int(commit_before_exit) + + +def test_conflicting_payload_under_a_committed_key_is_refused(mini_repo, tmp_path): + material_fixture(mini_repo) + first = task_ir.parse_spec(_spec()) + directory = tmp_path / "first" + task_ir.preview_task(mini_repo, first, directory, run_reads=_run_reads) + committed = task_ir.apply_task(mini_repo, first, directory) + assert committed["status"] == "committed" + changed = _spec() + changed["operations"][2]["set_fields"] = {"angle": "Conflicting new intent"} + second = task_ir.parse_spec(changed) + other = tmp_path / "second" + assert task_ir.preview_task(mini_repo, second, other, run_reads=_run_reads)["status"] == "needs_review" + after_commit = canonical_fingerprint(mini_repo) + result = task_ir.apply_task(mini_repo, second, other) + assert result["status"] == "refused" + assert len(_receipts(mini_repo)) == 1 + assert canonical_fingerprint(mini_repo) == after_commit + + +def test_operator_seal_does_not_claim_authenticated_human_approval(mini_repo, tmp_path): + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + directory = tmp_path / "evidence" + report = task_ir.preview_task(mini_repo, spec, directory, run_reads=_run_reads) + envelope = json.loads(next(directory.glob("*.envelope.json")).read_text()) + assert envelope["channel"] == "operator" + assert envelope["approval"]["kind"] == "operator-approval" + assert envelope["session_id"] == spec.session_id + assert report["operations"][-1]["outputs"]["changes"] == spec.operations[-1].set_fields + assert "direct-user-gesture" not in json.dumps(report) + assert "human_approved" not in json.dumps(report) + assert not _receipts(mini_repo) + + +def test_truncated_material_discovery_retains_expansion_and_cannot_bind(mini_repo, tmp_path): + repo, _, smid = material_fixture(mini_repo) + data = copy.deepcopy(repo.study_maps[smid].data) + data["stages"][0]["resources"][0]["angle_detail"] = "Long authored explanation. " * 1000 + write_yaml(repo.study_maps[smid].path, data) + spec = task_ir.parse_spec(_spec()) + before = canonical_fingerprint(mini_repo) + report = task_ir.preview_task(mini_repo, spec, tmp_path / "evidence", run_reads=_run_reads) + assert report["status"] == "partial" + material = report["operations"][1]["outputs"] + assert material["complete"] is False + assert material["route_id"] is None + assert material["materials"]["_truncated_bytes"] > 8000 + expanded = subprocess.run(material["expand_argv"], capture_output=True, text=True, timeout=60) + assert expanded.returncode == 0, expanded.stdout + expanded.stderr + full = json.loads(expanded.stdout) + original = json.loads(full["steps"][0]["raw"]["stdout"]) + assert len(original["stage"]["resources"]) == 3 + assert original["stage"]["resources"][0]["route_id"] + assert canonical_fingerprint(mini_repo) == before + assert not _receipts(mini_repo) + + +@pytest.mark.parametrize("attempt", ["route-in-another-unit", "stage-in-another-unit"]) +def test_unit_and_stage_scope_mismatches_refuse(mini_repo, tmp_path, attempt): + from learning_os.derive.validation import validate + + repo, route_id, _ = material_fixture(mini_repo) + other_unit = copy.deepcopy(repo.units["unit-demo-l01"].data) + other_unit.update(id="unit-demo-l02", title="Another unit", order=2, + status="archived", source_selections=[]) + other_unit.pop("current_study_map") + path = mini_repo / "curriculum/modules/module-demo/module.yaml" + module = yaml.safe_load(path.read_text()) + module["unit_order"].append("unit-demo-l02") + write_yaml(path, module) + write_yaml(path.parent / "units/unit-demo-l02/unit.yaml", other_unit) + assert not [issue for issue in validate(load_repo(mini_repo)) if issue.severity == "error"] + if attempt == "route-in-another-unit": + operations = [{"id": "patch", "op": "route.patch", "unit_id": "unit-demo-l02", + "route_id": route_id, "set_fields": {"angle": "Cross-unit attempt"}}] + else: + operations = [{"id": "find", "op": "material.find", "unit_id": "unit-demo-l02", "stage_id": "stage-demo"}, + {"id": "patch", "op": "route.patch", "unit_id": "unit-demo-l01", + "route_ref": {"from_operation": "find", "field": "route_id"}, + "set_fields": {"angle": "Cross-stage attempt"}}] + spec = task_ir.parse_spec({"schema_version": 1, "request_id": "scope-test", "operations": operations}) + before = canonical_fingerprint(mini_repo) + report = task_ir.preview_task(mini_repo, spec, tmp_path / "evidence", run_reads=_run_reads) + assert report["status"] == "partial" + assert report["operations"][-1]["status"] == "refused" + assert canonical_fingerprint(mini_repo) == before + assert not _receipts(mini_repo) + + +def test_read_epoch_change_refuses_instead_of_attaching_a_fresh_snapshot(mini_repo, tmp_path): + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + + def changing_reads(plan, root): + report = _run_reads(plan, root) + path = root / "knowledge/notes/mathematics/note-demo.md" + path.write_text(path.read_text() + "\nSynthetic concurrent change.\n") + return report + + with pytest.raises(task_ir.TaskSpecError, match="snapshot changed") as exc: + task_ir.preview_task(mini_repo, spec, tmp_path / "evidence", run_reads=changing_reads) + assert exc.value.code == "TASK_IR_STALE_READ" + assert not _receipts(mini_repo) + + +@pytest.mark.parametrize("field,value", [ + ("channel", "ui"), ("session_id", "another-session"), + ("request_id", "another-request"), ("idempotency_key", "another-key"), + ("approval", {"kind": "direct-user-gesture"}), +]) +def test_rehashed_envelope_cannot_change_scope_or_claim_human_authority(mini_repo, tmp_path, field, value): + from learning_os.store.contracts.gateway import intent_sha256 + + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + directory = tmp_path / "evidence" + task_ir.preview_task(mini_repo, spec, directory, run_reads=_run_reads) + path = next(directory.glob("*.envelope.json")) + envelope = json.loads(path.read_text()) + envelope[field] = value + envelope["approval"]["subject_sha256"] = intent_sha256(envelope) + path.write_text(json.dumps(envelope)) + state_path, state = _state(directory) + state["envelope_sha256"] = "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest() + state_path.write_text(json.dumps(state)) + with pytest.raises(task_ir.TaskSpecError): + task_ir.apply_task(mini_repo, spec, directory) + assert not _receipts(mini_repo) + + +@pytest.mark.parametrize("field,value", [("schema", 2), ("ok", False), ("capability", "capture.create")]) +def test_rehashed_invalid_check_record_is_still_refused(mini_repo, tmp_path, field, value): + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + directory = tmp_path / "evidence" + task_ir.preview_task(mini_repo, spec, directory, run_reads=_run_reads) + state_path, state = _state(directory) + check_path = Path(state["check_record"]) + check = json.loads(check_path.read_text()) + check[field] = value + check_path.write_text(json.dumps(check)) + state["check_sha256"] = "sha256:" + hashlib.sha256(check_path.read_bytes()).hexdigest() + state_path.write_text(json.dumps(state)) + with pytest.raises(task_ir.TaskSpecError): + task_ir.apply_task(mini_repo, spec, directory) + assert not _receipts(mini_repo) + + +def test_missing_original_after_interruption_cannot_be_resealed(mini_repo, tmp_path, monkeypatch): + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + directory = tmp_path / "evidence" + task_ir.preview_task(mini_repo, spec, directory, run_reads=_run_reads) + state_path, state = _state(directory) + state["status"] = "dispatched" + state_path.write_text(json.dumps(state)) + next(directory.glob("*.envelope.json")).unlink() + + def forbidden(*args): + pytest.fail("an interrupted request must never be resealed or blindly submitted") + + monkeypatch.setattr(task_ir, "_run_seal", forbidden) + result = task_ir.apply_task(mini_repo, spec, directory, seal_submit=forbidden) + assert result["status"] == "uncertain" + assert result["operations"][-1]["status"] == "uncertain" + assert not _receipts(mini_repo) + + +def test_success_without_real_commit_evidence_remains_uncertain(mini_repo, tmp_path): + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + directory = tmp_path / "evidence" + task_ir.preview_task(mini_repo, spec, directory, run_reads=_run_reads) + report = task_ir.apply_task(mini_repo, spec, directory, + seal_submit=lambda root, content: (0, '{"receipt_path":"invented.yaml"}', "")) + assert report["status"] == "uncertain" + assert "receipt" not in report + assert task_ir.apply_task(mini_repo, spec, directory)["status"] == "uncertain" + assert not _receipts(mini_repo) + + +def test_request_id_containing_check_suffix_keeps_one_artifact_stem(mini_repo, tmp_path): + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec(request_id="a.check.json")) + directory = tmp_path / "evidence" + assert task_ir.preview_task(mini_repo, spec, directory, run_reads=_run_reads)["status"] == "needs_review" + assert (directory / "task-a.check.json-patch-route.envelope.json").is_file() + assert task_ir.apply_task(mini_repo, spec, directory)["status"] == "committed" + assert len(_receipts(mini_repo)) == 1 + + +def test_unknown_unit_cannot_spoof_an_unchanged_result_through_error_text(mini_repo, tmp_path): + _, route_id, _ = material_fixture(mini_repo) + spec = task_ir.parse_spec({"schema_version": 1, "request_id": "spoof-no-change", "operations": [ + {"id": "patch", "op": "route.patch", "unit_id": "unit-no material changes to apply", + "route_id": route_id, "set_fields": {"angle": "Not a valid scoped edit"}}]}) + before = canonical_fingerprint(mini_repo) + report = task_ir.preview_task(mini_repo, spec, tmp_path / "evidence", run_reads=_run_reads) + assert report["status"] == "partial" + assert report["operations"][0]["status"] == "refused" + assert canonical_fingerprint(mini_repo) == before + assert not _receipts(mini_repo) + + +@pytest.mark.parametrize("destination", ["check", "temporary"]) +def test_hard_linked_output_cannot_overwrite_canonical_data(mini_repo, tmp_path, destination): + material_fixture(mini_repo) + spec = task_ir.parse_spec(_spec()) + directory = tmp_path / "evidence" + directory.mkdir() + stem = hashlib.sha256(spec.request_id.encode()).hexdigest()[:32] + path = (directory / f"task-{spec.request_id}-patch-route.check.json" if destination == "check" + else directory / f".task-{stem}.json.tmp") + canonical = mini_repo / "sources/sources.yaml" + original = canonical.read_bytes() + os.link(canonical, path) + before = canonical_fingerprint(mini_repo) + with pytest.raises(task_ir.TaskSpecError, match="hard links"): + task_ir.preview_task(mini_repo, spec, directory, run_reads=_run_reads) + assert canonical.read_bytes() == original + assert canonical_fingerprint(mini_repo) == before + assert not _receipts(mini_repo) diff --git a/tools/agent_run.py b/tools/agent_run.py index 98ee30a..02a4cdd 100644 --- a/tools/agent_run.py +++ b/tools/agent_run.py @@ -34,7 +34,7 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("--task", default=None, help="Task IR P1 spec JSON file (experimental; alternative to a plan)") parser.add_argument("--task-phase", choices=("preview", "apply"), default="preview", - help="Task IR phase: preview seals the reviewed check only; apply submits it once") + help="Task IR phase: preview retains the checked envelope; apply submits or reconciles it") parser.add_argument("--evidence-dir", default=None, help="Task IR run-evidence directory (outside the data root)") return parser diff --git a/tools/learning_os/app/task_ir.py b/tools/learning_os/app/task_ir.py index a3c95a7..32ad7a6 100644 --- a/tools/learning_os/app/task_ir.py +++ b/tools/learning_os/app/task_ir.py @@ -34,8 +34,8 @@ operation's idempotency key and gateway request ID are derived once at preview, persisted in the run evidence keyed by ``(request_id, operation_id)``, and looked up on every retry: apply resubmits through the -saved check record and the gateway's kept envelope, never resealing against -a fresh snapshot, and an uncertain dispatch is never redispatched. +saved original envelope bytes. The existing gateway replay-only path +reconciles uncertain dispatches without resealing or dispatching another write. """ from __future__ import annotations @@ -51,11 +51,15 @@ from typing import cast from learning_os.app import handlers as _handlers # noqa: F401 (registration) +from learning_os.app.agent_presentation import summarize from learning_os.app.agent_reporting import encode +from learning_os.app.gateway import context_from_v2, validate_capability_envelope from learning_os.app.registry import REGISTRY, OperationEffects from learning_os.derive.learning_runtime import requirement_id_for from learning_os.domain.failures import TransactionFailure from learning_os.domain.material_refs import PATCH_FIELDS +from learning_os.store.contracts.gateway import intent_sha256 +from learning_os.store.fingerprint import snapshot from learning_os.store.transactions import atomic_write_bytes TOOLS = Path(__file__).resolve().parents[2] @@ -180,22 +184,29 @@ def _operation(raw: object, earlier: dict[str, Operation]) -> Operation: if op_id in earlier: raise _fail(f"duplicate operation id: {op_id}") op = raw.get("op") - if op not in OUTPUTS: + if not isinstance(op, str) or op not in OUTPUTS: raise _fail(f"operation {op_id}: unknown op {op!r}") - allowed = {"id", "op"} + fields = { + "context.inspect": {"record_id", "unit_id", "stage_id", "route_id"}, + "material.find": {"unit_id", "stage_id"}, + "session.propose": {"unit_id", "stage_id", "context"}, + "route.patch": {"unit_id", "route_id", "route_ref", "set_fields", "claim_evidence"}, + } + extra = set(raw) - {"id", "op"} - fields[op] + if extra: + raise _fail(f"operation {op_id}: undeclared field(s): {', '.join(sorted(extra))}") if op == "context.inspect": - allowed |= {"record_id", "unit_id", "stage_id", "route_id"} + if sum(name in raw for name in fields[op]) != 1: + raise _fail(f"operation {op_id}: context.inspect needs exactly one record selector") record = raw.get("record_id") or raw.get("route_id") or raw.get("stage_id") or raw.get("unit_id") if not isinstance(record, str) or not record.strip(): raise _fail(f"operation {op_id}: context.inspect needs a known unit, stage, route, or record id") return Operation(id=op_id, op=op, record_id=record) if op == "material.find": - allowed |= {"unit_id", "stage_id"} return Operation(id=op_id, op=op, unit_id=_text(raw, "unit_id", op_id, required=True), stage_id=_text(raw, "stage_id", op_id)) if op == "session.propose": - allowed |= {"unit_id", "stage_id", "context"} context = raw.get("context", {}) if not isinstance(context, dict): raise _fail(f"operation {op_id}: context must be an object") @@ -203,7 +214,6 @@ def _operation(raw: object, earlier: dict[str, Operation]) -> Operation: unit_id=_text(raw, "unit_id", op_id, required=True), stage_id=_text(raw, "stage_id", op_id, required=True), context=dict(context)) - allowed |= {"unit_id", "route_id", "route_ref", "set_fields", "claim_evidence"} route_ref = _route_ref(raw, op_id, earlier) set_fields = raw.get("set_fields") if not isinstance(set_fields, dict) or not set_fields: @@ -216,9 +226,6 @@ def _operation(raw: object, earlier: dict[str, Operation]) -> Operation: evidence = raw.get("claim_evidence") if evidence is not None and not isinstance(evidence, dict): raise _fail(f"operation {op_id}: claim_evidence must be an object") - extra = set(raw) - allowed - if extra: - raise _fail(f"operation {op_id}: undeclared field(s): {', '.join(sorted(extra))}") return Operation(id=op_id, op=op, unit_id=_text(raw, "unit_id", op_id, required=True), route_ref=route_ref, set_fields=dict(set_fields), @@ -243,7 +250,7 @@ def parse_spec(raw: object) -> TaskSpec: extra = set(raw) - {"schema_version", "request_id", "session_id", "operations"} if extra: raise _fail(f"undeclared spec field(s): {', '.join(sorted(extra))}") - if raw.get("schema_version") != 1: + if type(raw.get("schema_version")) is not int or raw.get("schema_version") != 1: raise _fail("task spec schema_version must be 1") request_id = raw.get("request_id") if not isinstance(request_id, str) or not _ID.fullmatch(request_id) or len(request_id) > 128: @@ -309,7 +316,7 @@ def compile_spec(spec: TaskSpec) -> CompiledTask: and operation.route_ref.from_operation else []) if operation.op == "context.inspect": steps.append({"id": operation.id, "argv": ["inspect", str(operation.record_id)], - "summary": {"fields": ["id", "type"], "max_items": 5, "max_bytes": 2000}}) + "summary": {"max_items": 5, "max_bytes": 4000}}) elif operation.op == "material.find": argv = ["plan-edit-context", str(operation.unit_id)] fields = ["snapshot_id", "routes"] @@ -323,8 +330,8 @@ def compile_spec(spec: TaskSpec) -> CompiledTask: steps.append({"id": operation.id, "argv": ["runtime-session", "--requirement", requirement, "--context-json", encode(operation.context)], - "summary": {"fields": ["snapshot_id", "contract"], - "max_items": 5, "max_bytes": 2000}}) + "summary": {"fields": ["schema_version", "snapshot_id", "contract", "session"], + "max_items": 5, "max_bytes": 8000}}) plan: dict[str, object] = {"schema_version": 2, "session_id": spec.session_id, "on_error": "stop", "steps": steps} return CompiledTask(plan=plan, write=write, @@ -341,7 +348,7 @@ def resolve_route(compiled: CompiledTask, outputs: dict[str, dict[str, object]]) return write.route_ref.literal source = outputs.get(str(write.route_ref.from_operation), {}) route_id = source.get("route_id") - if not isinstance(route_id, str) or not route_id: + if source.get("complete") is not True or not isinstance(route_id, str) or not route_id: raise _fail("material.find did not resolve to exactly one route; refusing to guess", "TASK_IR_UNRESOLVED_ROUTE") return route_id @@ -356,8 +363,8 @@ def write_payload(write: Operation, route_id: str) -> dict[str, object]: ReadRunner = Callable[[dict[str, object], Path], dict[str, object]] -SealCheck = Callable[[Path, dict[str, object], str, Path], tuple[int, str, str]] -SealSubmit = Callable[[Path, Path, dict[str, object], str, str], tuple[int, str, str]] +SealCheck = Callable[[Path, dict[str, object], str, Path, str], tuple[int, str, str]] +SealSubmit = Callable[[Path, bytes], tuple[int, str, str]] def _run_seal(argv: list[str]) -> tuple[int, str, str]: @@ -366,18 +373,35 @@ def _run_seal(argv: list[str]) -> tuple[int, str, str]: def default_seal_check(root: Path, payload: dict[str, object], key: str, - record_path: Path) -> tuple[int, str, str]: - return _run_seal([sys.executable, str(TOOLS / "seal_envelope.py"), "--root", str(root), - "--capability", "route.patch", "--payload", encode(payload), - "--key", key, "--guards", "auto", "--check-out", str(record_path)]) + record_path: Path, session_id: str) -> tuple[int, str, str]: + common = [sys.executable, str(TOOLS / "seal_envelope.py"), "--root", str(root), + "--capability", "route.patch", "--key", key, "--request-id", f"request-{key}", + "--session-id", session_id] + # The declared check path produces structured empty-write evidence for a + # set-like no-op; error prose must never classify a successful outcome. + code, out, err = _run_seal([*common, "--payload", encode({**payload, "check": True}), + "--guards", "auto", "--check-out", str(record_path)]) + if code != 0: + return code, out, err + return _run_seal([*common, "--payload", encode(payload), "--guards-from", str(record_path)]) + +def _run_envelope(root: Path, content: bytes, *, replay_only: bool) -> tuple[int, str, str]: + # Submit the checked byte buffer, never reopen a caller-changeable path. + argv = [sys.executable, str(TOOLS / "los.py"), "--root", str(root), + "capability", "route.patch", "--payload-file", "-"] + if replay_only: + argv.append("--replay-only") + proc = subprocess.run(argv, input=content, capture_output=True, timeout=600) + return proc.returncode, proc.stdout.decode("utf-8"), proc.stderr.decode("utf-8") -def default_seal_submit(root: Path, record_path: Path, payload: dict[str, object], - key: str, request_id: str) -> tuple[int, str, str]: - return _run_seal([sys.executable, str(TOOLS / "seal_envelope.py"), "--root", str(root), - "--capability", "route.patch", "--payload", encode(payload), - "--key", key, "--request-id", request_id, - "--guards-from", str(record_path), "--submit"]) + +def default_seal_submit(root: Path, content: bytes) -> tuple[int, str, str]: + return _run_envelope(root, content, replay_only=False) + + +def default_seal_replay(root: Path, content: bytes) -> tuple[int, str, str]: + return _run_envelope(root, content, replay_only=True) def _evidence_path(evidence_dir: Path, request_id: str) -> Path: @@ -398,20 +422,85 @@ def _check_evidence_path(path: Path) -> None: raise _fail(f"evidence path traverses a symlink: {path}", "TASK_IR_EVIDENCE") if path.exists() and not path.is_file(): raise _fail(f"evidence path is not a regular file: {path}", "TASK_IR_EVIDENCE") + if path.exists() and path.stat().st_nlink != 1: + raise _fail(f"evidence path has multiple hard links: {path}", "TASK_IR_EVIDENCE") -def _save_evidence(path: Path, evidence: dict[str, object]) -> None: +def _save_bytes(path: Path, content: bytes) -> None: # As in agent_reporting, this is private recovery evidence, not a # canonical transaction or disposable cache. Reuse the shared persistence # barrier so dispatch cannot precede a durable checkpoint. _check_evidence_path(path) _check_evidence_path(path.with_name(f".{path.name}.tmp")) try: - atomic_write_bytes(path, (encode(evidence) + "\n").encode("utf-8")) + atomic_write_bytes(path, content) except TransactionFailure as exc: raise _fail(f"run evidence could not be persisted: {exc}", "TASK_IR_EVIDENCE") from exc +def _save_evidence(path: Path, evidence: dict[str, object]) -> None: + _save_bytes(path, (encode(evidence) + "\n").encode("utf-8")) + + +def _digest(content: bytes) -> str: + return "sha256:" + hashlib.sha256(content).hexdigest() + + +def _read_bound(path: Path, digest: object) -> bytes: + _check_evidence_path(path) + try: + content = path.read_bytes() + except OSError as exc: + raise _fail(f"review artifact is missing or unreadable: {path}", "TASK_IR_EVIDENCE") from exc + if digest != _digest(content): + raise _fail(f"review artifact changed: {path}", "TASK_IR_EVIDENCE") + return content + + +def _bound_submission(root: Path, spec: TaskSpec, evidence_dir: Path, + evidence: dict[str, object]) -> bytes: + """Bind the existing check and GatewayEnvelopeV2 to this exact task.""" + compiled = compile_spec(spec) + write = compiled.write + if write is None: + raise _fail("preview has no writable operation", "TASK_IR_EVIDENCE") + outputs = evidence.get("outputs") + if not isinstance(outputs, dict): + raise _fail("preview outputs are missing", "TASK_IR_EVIDENCE") + route_id = resolve_route(compiled, outputs) + payload = write_payload(write, route_id) + key = operation_key(spec.request_id, write.id) + if (evidence.get("data_root") != str(root.resolve()) or evidence.get("key") != key + or evidence.get("gateway_request_id") != f"request-{key}" + or evidence.get("payload") != payload or evidence.get("route_id") != route_id): + raise _fail("preview scope, payload or identity differs from the task", "TASK_IR_EVIDENCE") + stem = f"task-{spec.request_id}-{write.id}" + check_path = evidence_dir / f"{stem}.check.json" + _check_evidence_path(check_path) + if Path(str(evidence.get("check_record"))).resolve() != check_path: + raise _fail("check record must stay in the evidence directory", "TASK_IR_EVIDENCE") + check_bytes = _read_bound(check_path, evidence.get("check_sha256")) + content = _read_bound(evidence_dir / f"{stem}.envelope.json", evidence.get("envelope_sha256")) + try: + check, envelope = json.loads(check_bytes), json.loads(content) + if not isinstance(check, dict) or not isinstance(envelope, dict): + raise ValueError("review artifacts must be objects") + validate_capability_envelope(root, envelope, kind="request") + context_from_v2(envelope) + if (check.get("schema") != 1 or check.get("ok") is not True + or check.get("capability") != "route.patch" or check.get("payload") != payload + or envelope.get("payload") != payload or envelope.get("capability") != "route.patch" + or envelope.get("channel") != "operator" or envelope.get("session_id") != spec.session_id + or envelope.get("request_id") != f"request-{key}" or envelope.get("idempotency_key") != key + or envelope.get("approval", {}).get("kind") != "operator-approval" + or envelope.get("expected_snapshot") != check.get("expected_snapshot") + or envelope.get("expected_revisions") != check.get("expected_revisions")): + raise ValueError("check/envelope scope, payload, guards or identity differs") + except Exception as exc: + raise _fail(f"invalid reviewed artifacts: {exc}", "TASK_IR_EVIDENCE") from exc + return content + + def _load_evidence(path: Path, spec: TaskSpec) -> dict[str, object] | None: _check_evidence_path(path) if not path.is_file(): @@ -427,7 +516,15 @@ def _load_evidence(path: Path, spec: TaskSpec) -> dict[str, object] | None: return evidence -def _outputs_from_report(spec: TaskSpec, report: dict[str, object]) -> dict[str, dict[str, object]]: +def _incomplete(value: object) -> bool: + if isinstance(value, dict): + return any(key in value for key in ("_truncated_bytes", "_omitted", "_missing_fields")) \ + or any(_incomplete(item) for item in value.values()) + return isinstance(value, list) and any(_incomplete(item) for item in value) + + +def _outputs_from_report(spec: TaskSpec, report: dict[str, object], + snapshot_id: str) -> dict[str, dict[str, object]]: raw_steps = report.get("steps") rows: dict[str, dict[str, object]] = {} if isinstance(raw_steps, list): @@ -441,10 +538,19 @@ def _outputs_from_report(spec: TaskSpec, report: dict[str, object]) -> dict[str, row: dict[str, object] = rows.get(operation.id, {}) raw_output = row.get("output") output: dict[str, object] = raw_output if isinstance(raw_output, dict) else {} + observed = output.get("snapshot_id") + if observed is not None and observed != snapshot_id: + raise _fail("read result has a different snapshot; preview again", "TASK_IR_STALE_READ") + common: dict[str, object] = {"snapshot_id": snapshot_id, "status": row.get("status")} + if row.get("error") is not None: + common["error"] = row["error"] + if report.get("full_report"): + common["expand_argv"] = [sys.executable, str(TOOLS / "agent_run.py"), + "--expand-report", report["full_report"], "--step", operation.id] + if row.get("output_omitted") is True: + common["output_omitted"] = True if operation.op == "context.inspect": - outputs[operation.id] = {"record_id": operation.record_id, - "snapshot_id": output.get("snapshot_id"), - "status": row.get("status")} + outputs[operation.id] = {**common, "record_id": operation.record_id, "context": output} elif operation.op == "material.find": route_ids: list[str] = [] stage = output.get("stage") @@ -456,30 +562,19 @@ def _outputs_from_report(spec: TaskSpec, report: dict[str, object]) -> dict[str, route_ids = [str(item["id"]) for item in raw_routes if isinstance(item, dict) and item.get("id")] route_ids = list(dict.fromkeys(route_ids)) + complete = bool(output) and not _incomplete(output) and row.get("output_omitted") is not True outputs[operation.id] = { - "route_ids": route_ids, - "route_id": route_ids[0] if len(route_ids) == 1 else None, - "snapshot_id": output.get("snapshot_id"), "status": row.get("status")} + **common, "route_ids": route_ids, "complete": complete, + "route_id": route_ids[0] if complete and len(route_ids) == 1 else None, + "materials": output} else: outputs[operation.id] = { + **common, "requirement_id": requirement_id_for(str(operation.unit_id), str(operation.stage_id)), - "snapshot_id": output.get("snapshot_id"), "status": row.get("status")} + "proposal": output} return outputs -def _record_changes(record: dict[str, object]) -> dict[str, object] | None: - """The previewed per-field changes from a seal check record, if carried.""" - for value in record.values(): - if isinstance(value, dict): - changes = value.get("changes") - if isinstance(changes, dict): - return changes - nested = _record_changes(value) - if nested is not None: - return nested - return None - - def preview_task(root: Path, spec: TaskSpec, evidence_dir: Path, *, run_reads: ReadRunner, seal_check: SealCheck = default_seal_check) -> dict[str, object]: @@ -488,12 +583,16 @@ def preview_task(root: Path, spec: TaskSpec, evidence_dir: Path, *, evidence_dir.mkdir(parents=True, exist_ok=True, mode=0o700) path = _evidence_path(evidence_dir, spec.request_id) prior = _load_evidence(path, spec) - if prior is not None and prior.get("status") in ("committed", "uncertain"): + if prior is not None and prior.get("status") in ( + "needs_review", "committed", "uncertain", "dispatched", "refused", "unchanged"): return _report(spec, prior, resumed=True) compiled = compile_spec(spec) empty_report: dict[str, object] = {"steps": [], "totals": {"ok": 0, "steps": 0}} + read_snapshot = snapshot(root) report = run_reads(compiled.plan, root) if compiled.read_ids else empty_report - outputs = _outputs_from_report(spec, report) + if snapshot(root) != read_snapshot: + raise _fail("snapshot changed during task reads; preview again", "TASK_IR_STALE_READ") + outputs = _outputs_from_report(spec, report, read_snapshot) operations: list[dict[str, object]] = [] for operation in spec.operations: if operation.op == "route.patch": @@ -505,7 +604,9 @@ def preview_task(root: Path, spec: TaskSpec, evidence_dir: Path, *, if k != "status"}}) evidence: dict[str, object] = { "request_id": spec.request_id, "spec_digest": spec_digest(spec), - "status": "previewed", "operations": operations, "outputs": outputs, + "data_root": str(root.resolve()), + "status": "ok" if all(row["status"] == "ok" for row in operations) else "refused", + "operations": operations, "outputs": outputs, } if compiled.write is not None: write = compiled.write @@ -526,101 +627,125 @@ def preview_task(root: Path, spec: TaskSpec, evidence_dir: Path, *, key = operation_key(spec.request_id, write.id) record_path = evidence_dir / f"task-{spec.request_id}-{write.id}.check.json" _check_evidence_path(record_path) - code, _out, err = seal_check(root, payload, key, record_path) - if code != 0 and "no material changes to apply" in err: - # The existing machinery itself refuses a no-change patch - # before any write; that refusal is the unchanged proof. - evidence.update({"key": key, "payload": payload, "route_id": route_id}) - operations.append({"id": write.id, "op": write.op, - "status": "unchanged", - "outputs": {"route_id": route_id}}) - evidence["status"] = "unchanged" - elif code != 0 or not record_path.is_file(): + code, out, err = seal_check(root, payload, key, record_path, spec.session_id) + if code != 0 or not record_path.is_file(): operations.append({"id": write.id, "op": write.op, "status": "refused", "outputs": {"route_id": route_id}, "error": (err.strip().splitlines() or ["preview check failed"])[-1]}) evidence["status"] = "partial" else: - record = json.loads(record_path.read_text(encoding="utf-8")) - changes = _record_changes(record) - unchanged = changes is not None and not changes + record_bytes = record_path.read_bytes() + record = json.loads(record_bytes) + if record.get("expected_snapshot") != read_snapshot: + raise _fail("snapshot changed before the patch check; preview again", + "TASK_IR_STALE_READ") + unchanged = record.get("writes") == [] and record.get("deletes") == [] evidence.update({"key": key, "gateway_request_id": f"request-{key}", "payload": payload, "check_record": str(record_path), - "route_id": route_id}) + "route_id": route_id, "check_sha256": _digest(record_bytes), + "envelope_sha256": _digest(out.encode("utf-8"))}) + _save_bytes(record_path, record_bytes) + _save_bytes(evidence_dir / f"task-{spec.request_id}-{write.id}.envelope.json", + out.encode("utf-8")) + original = _bound_submission(root, spec, evidence_dir, evidence) + envelope = json.loads(original) operations.append({"id": write.id, "op": write.op, "status": "unchanged" if unchanged else "needs_review", - "outputs": {"route_id": route_id}}) + "outputs": {"route_id": route_id, + "changes": json.loads(summarize(write.set_fields, max_bytes=4000)), + "snapshot_id": envelope["expected_snapshot"], + "expected_revisions": envelope["expected_revisions"], + "intent_sha256": intent_sha256(envelope), + "check_record": str(record_path)}}) evidence["status"] = "unchanged" if unchanged else "needs_review" _save_evidence(path, evidence) return _report(spec, evidence) -def _committed_receipt(root: Path, key: str) -> dict[str, object] | None: - from learning_os.store.evidence import load_idempotency_entries +def _committed_receipt(root: Path, content: bytes) -> dict[str, object] | None: + from learning_os.store.evidence import verify_committed_evidence - entry = load_idempotency_entries(root).get(key) - if entry is None: + request = context_from_v2(json.loads(content)) + proof = verify_committed_evidence(root, request, authority_root=TOOLS.parent) + if proof is None: return None - return {"transaction_id": entry.get("transaction_id"), - "receipt_path": entry.get("receipt_path")} + _receipt, row, _path = proof + return {"transaction_id": row["transaction_id"], "receipt_path": row["receipt_path"]} def apply_task(root: Path, spec: TaskSpec, evidence_dir: Path, *, - seal_submit: SealSubmit = default_seal_submit) -> dict[str, object]: - """Apply the previewed write once; retries look up, never redispatch blind.""" + seal_submit: SealSubmit = default_seal_submit, + seal_replay: SealSubmit = default_seal_replay) -> dict[str, object]: + """Submit original reviewed bytes once; reconcile only through replay-only.""" evidence_dir = _evidence_directory(root, evidence_dir) path = _evidence_path(evidence_dir, spec.request_id) evidence = _load_evidence(path, spec) if evidence is None: raise _fail("no preview evidence for this request; run preview first", "TASK_IR_NO_PREVIEW") - if evidence.get("status") in ("committed", "uncertain", "unchanged"): + status = evidence.get("status") + recovering = status in ("committed", "uncertain", "dispatched") + if not recovering and status != "needs_review": return _report(spec, evidence, resumed=True) - if evidence.get("status") != "needs_review": - return _report(spec, evidence, resumed=True) - compiled = compile_spec(spec) - assert compiled.write is not None # needs_review is only set with a write - key = str(evidence["key"]) - record_path = Path(str(evidence["check_record"])) - expected_record = evidence_dir / f"task-{spec.request_id}-{compiled.write.id}.check.json" - _check_evidence_path(expected_record) - if record_path.resolve() != expected_record: - raise _fail("check record must stay in the evidence directory", "TASK_IR_EVIDENCE") - record_path = expected_record - payload = evidence["payload"] - if not isinstance(payload, dict) or not record_path.is_file(): - raise _fail("preview evidence is incomplete", "TASK_IR_EVIDENCE") - evidence["status"] = "dispatched" - _save_evidence(path, evidence) - - def settle(status: str, **extra: object) -> dict[str, object]: + def settle(next_status: str, **extra: object) -> dict[str, object]: raw_operations = evidence.get("operations") operations: list[object] = [dict(row) if isinstance(row, dict) else row for row in raw_operations] \ if isinstance(raw_operations, list) else [] for row in operations: if isinstance(row, dict) and row.get("op") == "route.patch": - row["status"] = status - evidence.update({"status": status, "operations": operations, **extra}) + row["status"] = next_status + raw_outputs = row.get("outputs") + outputs = dict(raw_outputs) if isinstance(raw_outputs, dict) else {} + outputs.pop("transaction_id", None) + outputs.pop("receipt_path", None) + receipt = extra.get("receipt") + if next_status == "committed" and isinstance(receipt, dict): + outputs.update({name: receipt[name] for name in ("transaction_id", "receipt_path")}) + row["outputs"] = outputs + evidence.pop("error", None) + evidence.pop("receipt", None) + evidence.update({"status": next_status, "operations": operations, **extra}) + _save_evidence(path, evidence) + return _report(spec, evidence, resumed=recovering) + + # Legacy interrupted evidence cannot establish its original bytes. Never + # replace it with a fresh seal, even when a receipt might exist. + try: + original = _bound_submission(root, spec, evidence_dir, evidence) + except TaskSpecError: + if recovering: + return settle("uncertain", error="original reviewed artifacts are unavailable or changed; retain evidence for reconciliation") + raise + + if not recovering: + evidence["status"] = "dispatched" _save_evidence(path, evidence) - return _report(spec, evidence) try: - code, out, err = seal_submit(root, record_path, payload, key, - str(evidence["gateway_request_id"])) - except Exception as exc: # dispatch may have crossed the commit boundary - return settle("uncertain", error=f"dispatch interrupted: {exc}") - if code == 0: + code, out, err = (seal_replay if recovering else seal_submit)(root, original) + except Exception as exc: # an attempted submission may already have committed + return settle("uncertain", error=f"dispatch/reconciliation interrupted: {exc}") + try: result = json.loads(out) if out.strip() else {} - receipt = {"transaction_id": result.get("transaction_id"), - "receipt_path": result.get("receipt_path"), - "replayed": result.get("replayed") is True} - if not receipt["receipt_path"] or not (root / str(receipt["receipt_path"])).is_file(): - return settle("uncertain", - error="gateway reported success without a receipt on disk") + if not isinstance(result, dict): + result = {} + except ValueError: + result = {} + error = result.get("error") + error_code = error.get("code") if isinstance(error, dict) else None + try: + receipt = _committed_receipt(root, original) + except Exception as exc: + if not recovering and error_code == "IDEMPOTENCY_CONFLICT": + return settle("refused", error="gateway refused conflicting idempotency intent") + return settle("uncertain", error=f"commit evidence could not be verified: {exc}") + if receipt is not None: + # A replay-only transport failure is not permission to infer that the + # call succeeded; the strict shared verifier still proves the commit. + receipt["replayed"] = recovering or result.get("replayed") is True return settle("committed", receipt=receipt) - recovered = _committed_receipt(root, key) - if recovered is not None: - return settle("committed", receipt=recovered) - detail = (err.strip().splitlines() or [out.strip() or "gateway refused the write"])[-1] + detail = (err.strip().splitlines() or [out.strip() or "gateway returned no verified receipt"])[-1] + if recovering or code == 0 or not error_code or error_code == "UNCONFIRMED": + return settle("uncertain", error=detail) return settle("refused", error=detail) diff --git a/work/complaints/2026-10-27-task-ir-evidence-persistence.md b/work/complaints/2026-10-27-task-ir-evidence-persistence.md index 7cc4b9e..1a8fe30 100644 --- a/work/complaints/2026-10-27-task-ir-evidence-persistence.md +++ b/work/complaints/2026-10-27-task-ir-evidence-persistence.md @@ -7,4 +7,4 @@ - Smallest repair: reuse `store.transactions.atomic_write_bytes`, already used by `app.agent_reporting`, and reject redirected evidence/check destinations before use. The repair adds no ratchet or effect-classification exemption. - Verification: `tests/test_task_ir.py` plus `tests/test_arch_writes.py` passed 39 cases, including six preview/apply symlink cases, a directory redirected into the data root, a failed checkpoint publication that prevented submission and retained the previous preview, and compatibility with previously persisted relative check-record paths. The existing file/directory synchronization test also passed. This does not establish protection against concurrent hostile directory replacement or complete P1 recovery acceptance. - Evidence: `../repair-baseline.log`, `../repair-focused-final.log`, `../repair-durability.log` in the experiment workbench; reproduction commands are retained in `../REPAIR-PLAN.md`. -- GitHub: not filed. Automatic approval review rejected publishing the first prepared issue to `Aram164/LearningOS`, stating that the specific contents and destination require explicit approval; the remaining issue submissions were halted. The exact ready-to-file body is this note, without this final status line. +- GitHub: https://github.com/Aram164/LearningOS/issues/65 diff --git a/work/complaints/2026-10-27-task-ir-false-unchanged.md b/work/complaints/2026-10-27-task-ir-false-unchanged.md new file mode 100644 index 0000000..3f8253c --- /dev/null +++ b/work/complaints/2026-10-27-task-ir-false-unchanged.md @@ -0,0 +1,9 @@ +# Task IR refusal text can spoof a successful unchanged result + +- Workflow: independent P1 acceptance review of route-patch preview classification. +- Tested state: Core `d707aaa` and the acceptance candidate, macOS/Python 3.14.4; no real learner data changed. +- Expected: an unknown unit must refuse; only verified set-like no-op evidence may produce `unchanged`. +- Observed: the reviewer used the unknown unit ID `unit-no material changes to apply` with a real synthetic route ID. The sealer's unknown-unit error included that text, so the adapter's substring check returned `unchanged` with success semantics. There was no valid scoped operation. +- Smallest repair: remove prose-based success classification. Run the existing explicit check path to obtain its structured write/delete lists; derive a route-patch no-op only from an admitted successful empty proposal. Retain the original check/seal/gateway mechanism, without generalizing no-op behavior to creates or appends. +- Acceptance evidence: `tests/test_task_ir.py::test_unknown_unit_cannot_spoof_an_unchanged_result_through_error_text` asserts refusal, unchanged canonical fingerprint and zero receipts; the legitimate unchanged-route test still passes. Final full-pair results are retained in the workbench acceptance report. +- GitHub: https://github.com/Aram164/LearningOS/issues/70 diff --git a/work/complaints/2026-10-27-task-ir-hardlink-preview.md b/work/complaints/2026-10-27-task-ir-hardlink-preview.md new file mode 100644 index 0000000..2d45002 --- /dev/null +++ b/work/complaints/2026-10-27-task-ir-hardlink-preview.md @@ -0,0 +1,9 @@ +# Task IR evidence hardlinks can redirect preview writes into canonical files + +- Workflow: independent P1 acceptance review of private evidence persistence. +- Tested state: Core `d707aaa` and the acceptance candidate, macOS/Python 3.14.4, disposable material fixtures only. +- Expected: preview cannot alter canonical files through its evidence/check output paths and creates no canonical receipt. +- Observed: the separate reviewer hard-linked `sources/sources.yaml` to the predictable Task IR check-record path. Preview followed the regular file, returned `needs_review`, changed canonical bytes/fingerprint, and produced zero receipts. The atomic sibling temporary path also accepts a hardlink unless checked before its write. Symlink-only checks do not detect these aliases. +- Smallest repair: refuse existing evidence/check/temp regular files whose link count differs from one before invoking any writer. Retain the existing shared atomic persistence primitive and external evidence directory. +- Acceptance evidence: `tests/test_task_ir.py::test_hard_linked_output_cannot_overwrite_canonical_data` covers check and temporary destinations and asserts unchanged canonical bytes/fingerprint and no receipts. Full Task IR candidate tests and final hosted pair results are recorded in the workbench acceptance report. This does not claim protection against hostile concurrent directory replacement. +- GitHub: https://github.com/Aram164/LearningOS/issues/65#issuecomment-6455314967 diff --git a/work/complaints/2026-10-27-task-ir-open-read-shapes.md b/work/complaints/2026-10-27-task-ir-open-read-shapes.md index f2cb613..5c71ccd 100644 --- a/work/complaints/2026-10-27-task-ir-open-read-shapes.md +++ b/work/complaints/2026-10-27-task-ir-open-read-shapes.md @@ -8,4 +8,6 @@ - Impact: a misspelled constraint can be silently ignored; this is a compile-contract failure, not evidence of a gateway authorization bypass. - Smallest improvement: apply the existing per-operation allowed-field check before every return. Add real-interface refusal cases for each read operation without creating another capability catalogue. - Acceptance: all three read operations reject unknown fields before dispatch, while valid existing specifications still compile deterministically. No repair for this gap is included in the immediate verification increment. -- GitHub: not filed. Automatic approval review rejected publishing the first prepared issue to `Aram164/LearningOS`, stating that the specific contents and destination require explicit approval; the remaining issue submissions were halted. The exact ready-to-file body is this note, without this final status line. +- GitHub: https://github.com/Aram164/LearningOS/issues/67 + +- Acceptance follow-up: repaired in the separate P1 acceptance increment. The candidate Task IR suite passed 73 cases, with an independent review reporting no unresolved blockers. Final head/pair/hosted evidence is retained in `LearningOS/workbench/task-ir-p1-experiment/P1-ACCEPTANCE-REPORT.md`; hosted acceptance is not inferred from the local run. diff --git a/work/complaints/2026-10-27-task-ir-read-results.md b/work/complaints/2026-10-27-task-ir-read-results.md index 34608ee..402f820 100644 --- a/work/complaints/2026-10-27-task-ir-read-results.md +++ b/work/complaints/2026-10-27-task-ir-read-results.md @@ -8,4 +8,6 @@ - Impact: callers treat a successful read-only task as failure, and inspecting context or proposing a session cannot deliver its central result through the public task report. - Smallest improvement: settle read-only tasks from their actual operation results and align the CLI mapping; carry bounded inspect/session projections and snapshots from the existing runner. Reuse existing full-report references for larger content. - Acceptance: real CLI tests distinguish successful and refused read-only tasks; a valid authored runtime target produces a snapshot-bound session contract without evidence/mastery claims or canonical changes. No new session entity or requirement store. This remains later P1 acceptance work. -- GitHub: not filed. Automatic approval review rejected publishing the first prepared issue to `Aram164/LearningOS`, stating that the specific contents and destination require explicit approval; the remaining issue submissions were halted. The exact ready-to-file body is this note, without this final status line. +- GitHub: https://github.com/Aram164/LearningOS/issues/69 + +- Acceptance follow-up: repaired in the separate P1 acceptance increment. The candidate Task IR suite passed 73 cases, with an independent review reporting no unresolved blockers. Final head/pair/hosted evidence is retained in `LearningOS/workbench/task-ir-p1-experiment/P1-ACCEPTANCE-REPORT.md`; hosted acceptance is not inferred from the local run. diff --git a/work/complaints/2026-10-27-task-ir-transcript-checksum.md b/work/complaints/2026-10-27-task-ir-transcript-checksum.md index 65e9265..0a316c9 100644 --- a/work/complaints/2026-10-27-task-ir-transcript-checksum.md +++ b/work/complaints/2026-10-27-task-ir-transcript-checksum.md @@ -7,4 +7,4 @@ - Smallest repair: retain strict mypy coverage and update only the transcript's expected pyproject hash. Do not weaken `tests/procedure_helpers/los-anchored-copy` or add wildcard output. - Acceptance: the exact hash matches `shasum -a 256 pyproject.toml`, and the complete procedure suite passes with its existing audit witnesses and expected failures. Final counts are retained in the workbench report. - Evidence: `../repair-baseline.log`, `../repair-procedures.log`, and `../REPAIR-REPORT.md` in the experiment workbench. -- GitHub: not filed. Automatic approval review rejected publishing the first prepared issue to `Aram164/LearningOS`, stating that the specific contents and destination require explicit approval; the remaining issue submissions were halted. The exact ready-to-file body is this note, without this final status line. +- GitHub: https://github.com/Aram164/LearningOS/issues/66 diff --git a/work/complaints/2026-10-27-task-ir-uncertain-reconciliation.md b/work/complaints/2026-10-27-task-ir-uncertain-reconciliation.md index c0c3f01..0f95211 100644 --- a/work/complaints/2026-10-27-task-ir-uncertain-reconciliation.md +++ b/work/complaints/2026-10-27-task-ir-uncertain-reconciliation.md @@ -8,4 +8,6 @@ - Impact: the adapter safely avoids blind redispatch but strands recoverable committed outcomes. Its current test proves non-redispatch, not completed recovery. - Smallest improvement: use the existing receipt/retained-envelope recovery mechanism for uncertain/dispatched states, preserve exact identity, and define explicit operator action when evidence cannot establish an outcome. No second journal or resealing protocol. - Acceptance: adapter-level lost-response, abrupt-exit, missing-receipt, conflicting-intent and exact-replay cases prove one transaction, one receipt, preserved identities, and truthful unresolved outcomes. This is deferred acceptance work. -- GitHub: not filed. Automatic approval review rejected publishing the first prepared issue to `Aram164/LearningOS`, stating that the specific contents and destination require explicit approval; the remaining issue submissions were halted. The exact ready-to-file body is this note, without this final status line. +- GitHub: https://github.com/Aram164/LearningOS/issues/68 + +- Acceptance follow-up: repaired in the separate P1 acceptance increment. The candidate Task IR suite passed 73 cases, with an independent review reporting no unresolved blockers. Final head/pair/hosted evidence is retained in `LearningOS/workbench/task-ir-p1-experiment/P1-ACCEPTANCE-REPORT.md`; hosted acceptance is not inferred from the local run. From e895758126e332f8a8a9d7b34d4f1916337cee45 Mon Sep 17 00:00:00 2001 From: Aram Aljanadi Date: Wed, 28 Oct 2026 00:00:23 +0100 Subject: [PATCH 5/5] Use canonical error codes in Task IR recovery Hosted A14 caught a literal IDEMPOTENCY_CONFLICT comparison. Import it and UNCONFIRMED from the existing domain vocabulary without changing recovery behavior or weakening the architecture ratchet. LOS-CI-Bootstrap: codex/task-ir-p1-experiment --- tools/learning_os/app/task_ir.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tools/learning_os/app/task_ir.py b/tools/learning_os/app/task_ir.py index 32ad7a6..20335db 100644 --- a/tools/learning_os/app/task_ir.py +++ b/tools/learning_os/app/task_ir.py @@ -56,6 +56,7 @@ from learning_os.app.gateway import context_from_v2, validate_capability_envelope from learning_os.app.registry import REGISTRY, OperationEffects from learning_os.derive.learning_runtime import requirement_id_for +from learning_os.domain.codes import IDEMPOTENCY_CONFLICT, UNCONFIRMED from learning_os.domain.failures import TransactionFailure from learning_os.domain.material_refs import PATCH_FIELDS from learning_os.store.contracts.gateway import intent_sha256 @@ -735,7 +736,7 @@ def settle(next_status: str, **extra: object) -> dict[str, object]: try: receipt = _committed_receipt(root, original) except Exception as exc: - if not recovering and error_code == "IDEMPOTENCY_CONFLICT": + if not recovering and error_code == IDEMPOTENCY_CONFLICT: return settle("refused", error="gateway refused conflicting idempotency intent") return settle("uncertain", error=f"commit evidence could not be verified: {exc}") if receipt is not None: @@ -744,7 +745,7 @@ def settle(next_status: str, **extra: object) -> dict[str, object]: receipt["replayed"] = recovering or result.get("replayed") is True return settle("committed", receipt=receipt) detail = (err.strip().splitlines() or [out.strip() or "gateway returned no verified receipt"])[-1] - if recovering or code == 0 or not error_code or error_code == "UNCONFIRMED": + if recovering or code == 0 or not error_code or error_code == UNCONFIRMED: return settle("uncertain", error=detail) return settle("refused", error=detail)