Declare effects and guard uncertain writes in agent plans

system/developer/components/operation-pipeline.md

@@ -47,6 +47,11 @@ Used by: `ai-delivery`, `cli-adapter`, `operation-handlers`, `plan-gate`.
 ## Invariants
 
 - Never dispatch an unadmitted write or bypass snapshot/replay checks. Pinning: `tests/test_ops_pipeline.py`.
+- Declare `OperationEffects` per operation independently of invocation kind. Pinning: `tests/test_agent_run.py`.
+  Only `READ` permits the runner to schedule a read. `DERIVED_WRITE` includes
+  optional report, cache, staging and export output; `CANONICAL_WRITE` includes
+  canonical records and Git/session mutations. The default is `UNKNOWN`, a
+  write barrier. This scheduling metadata grants no write authority.
 
 Admission lives in `system/contracts/capabilities.yaml`; `app/admission.py` constants
 `GESTURE_ALLOWLIST` and `UI_REVIEWED_ALLOWLIST` enforce it. The UI mirrors

system/procedures/agent-plans.md

@@ -52,6 +52,15 @@ The default worker count is one. Opt in to two through four workers only for
 measured workloads; process startup can dominate small plans. Read waves stop at
 every write or unknown-effect operation. `deadline_s` stops admission and
 cancellable reads, while an admitted commit is allowed to finish.
+Only operations explicitly declaring `read` effects in the executing Core are
+reads. Derived writes, canonical writes (including `session-end`), and unknown effects
+require a plan `session_id` and cannot run after an unresolved write, for both
+`argv` and `use` steps. Describing an operation remains a read.
+Every dispatched write starts with uncertain commit status, including porcelain.
+A receipt establishes commitment; an explicit gateway refusal can establish no
+commit. A porcelain error's generic `commit: none` cannot prove that it had no
+side effects. Without reliable evidence, later writes are skipped even under
+`on_error: continue`; independent reads can still run.
 
 `report_max_bytes` caps the entire stdout report. The `full_report` pointer
 leads to a private report outside canonical data, with the approved plan, sealed

tests/test_agent_run.py

@@ -679,6 +679,190 @@ def unnecessary(*_):
     assert report['totals']['ok'] == 1, report
 
 
+@pytest.mark.parametrize(("kind", "effects", "writer"), [
+    (None, "unknown", True), ("unknown", "unknown", True),
+    ("porcelain", "unknown", True), ("command", "canonical-write", True),
+    ("query", "unknown", True), ("tool", "unknown", True),
+    ("tool", "derived-write", True), ("query", "read", False),
+    ("tool", "read", False), ("porcelain", "read", False),
+])
+def test_write_classification_is_conservative_and_independent_of_task_resolution(mini_repo, monkeypatch, kind, effects, writer):
+    from dataclasses import replace
+
+    from learning_os.app.registry import REGISTRY, OperationEffects
+
+    executor = agent_run._cli_executor_factory(mini_repo.resolve())
+    original = REGISTRY["project.inspect"]
+    if kind is None:
+        monkeypatch.delitem(REGISTRY, "project.inspect")
+    else:
+        monkeypatch.setitem(REGISTRY, "project.inspect", replace(original, kind=kind, effects=OperationEffects(effects)))
+    direct = agent_run.build_index(mini_repo, executor, resolve_tasks=False)
+    expanded = agent_run.build_index(mini_repo, executor, resolve_tasks=True)
+    assert set(direct["write_verbs"]) == set(expanded["write_verbs"])
+    for index in (direct, expanded):
+        assert {"session-end", "generate", "ai-action-prepare", "ai-action-append-slices",
+                "ai-action-import-delivery"}.issubset(index["write_verbs"])
+        assert ("inspect" in index["write_verbs"]) is writer
+        assert "bootstrap" not in index["write_verbs"]
+        assert "capabilities" not in index["write_verbs"]
+
+
+def test_query_and_tool_effects_are_explicit_and_file_outputs_are_writers(mini_repo):
+    from learning_os.app.registry import REGISTRY, OperationEffects
+
+    executor = agent_run._cli_executor_factory(mini_repo.resolve())
+    assert not [op.name for op in REGISTRY.values()
+                if op.kind in {"query", "tool"} and op.effects == OperationEffects.UNKNOWN]
+    writers = set(agent_run.build_index(mini_repo, executor, resolve_tasks=False)["write_verbs"])
+    assert {"validate", "note-analysis-prepare", "backup-manifest", "backup-verify", "search", "dossier"} <= writers
+    assert REGISTRY["ai-action.delivery.validate"].effects == OperationEffects.READ
+    assert "ability-context" not in writers  # Explicit read despite porcelain invocation.
+    with pytest.raises(agent_run.AgentRunError) as error:
+        agent_run.run_plan({"steps": [{"id": "validation", "argv": ["validate"]}]}, mini_repo)
+    assert error.value.code == "SESSION_REQUIRED"
+
+
+@pytest.mark.parametrize("form", ["argv", "prefixed", "use"])
+@pytest.mark.parametrize("mixed", [False, True])
+def test_session_end_commit_push_requires_plan_session_before_dispatch(mini_repo, monkeypatch, form, mixed):
+    from types import SimpleNamespace
+
+    from learning_os.adapters import cli
+
+    calls = []
+    monkeypatch.setattr(cli, "invoke_outcome", lambda ns: (
+        calls.append(ns.command) or SimpleNamespace(exit_code=0, stdout='{"ok":true}', stderr="")))
+    args = ["--commit-message", "fixture commit", "--push"]
+    step = ({"id": "end", "use": "session.end", "args": args} if form == "use" else
+            {"id": "end", "argv": [*(["--lock-timeout", "1"] if form == "prefixed" else []), "session-end", *args]})
+    step["write"] = False  # A presentation hint cannot weaken registry effects.
+    steps = ([{"id": "read", "use": "inspect", "args": ["note-demo"]}] if mixed else []) + [step]
+    try:
+        report = agent_run.run_plan({"steps": steps}, mini_repo)
+    except agent_run.AgentRunError as exc:
+        assert exc.code == "SESSION_REQUIRED"
+    else:
+        assert report["steps"][-1]["error"]["code"] == "SESSION_REQUIRED", report
+    assert calls == []
+
+
+@pytest.mark.parametrize("use", [False, True])
+def test_uncertain_write_blocks_porcelain_but_allows_independent_read(mini_repo, monkeypatch, use):
+    from types import SimpleNamespace
+
+    from learning_os.adapters import cli
+
+    calls = []
+
+    def lost_write_response(argv, **kwargs):
+        assert any(str(arg).endswith("/seal_envelope.py") for arg in argv)
+        calls.append("seal")
+        return subprocess.CompletedProcess(argv, 1, "", "fixture lost write response")
+
+    def dispatch(namespace):
+        calls.append(namespace.command)
+        return SimpleNamespace(exit_code=0, stdout='{"ok":true}', stderr="")
+
+    monkeypatch.setattr(agent_run.subprocess, "run", lost_write_response)
+    monkeypatch.setattr(cli, "invoke_outcome", dispatch)
+    step = ({"id": "end", "use": "session-end", "args": ["--commit-message", "fixture", "--push"]}
+            if use else {"id": "end", "argv": ["session-end", "--commit-message", "fixture", "--push"]})
+    report = agent_run.run_plan({"session_id": "classification-fixture", "on_error": "continue", "steps": [
+        {"id": "write", "kind": "log", "text": "fixture", "key": "classification-fixture"},
+        step, {"id": "read", "argv": ["inspect", "note-demo"]},
+    ]}, mini_repo)
+    assert calls == ["seal", "inspect"], report
+    assert report["steps"][0]["commit_certainty"] == "uncertain"
+    assert report["steps"][1]["status"] == "skipped"
+    assert report["steps"][1]["error"]["code"] == "PLAN_STOPPED"
+    assert report["steps"][2]["status"] == "ok"
+
+
+def test_describing_porcelain_requires_no_write_session(mini_repo):
+    report = agent_run.run_plan({"steps": [
+        {"id": "describe", "use": "session-end", "describe": True},
+    ]}, mini_repo)
+    assert report["totals"]["ok"] == 1, report
+    assert report["steps"][0]["commit_certainty"] == "none"
+
+
+@pytest.mark.parametrize("use", [False, True])
+def test_session_commit_then_failed_push_blocks_later_writes(mini_repo, use):
+    from test_session_ledger import _git_init, _sealed_capture
+
+    from learning_os.store import session_ledger as session_store
+
+    session = "runner-partial-porcelain"
+    _git_init(mini_repo)
+    _sealed_capture(mini_repo, key="before-partial-porcelain", text="fixture capture", session_id=session)
+    execute = agent_run._cli_executor_factory(mini_repo.resolve(), session_id=session)
+    code, _, err = execute(["session-end", "--session-id", session], mini_repo)
+    assert code == 0, err
+    before = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=mini_repo, text=True)
+    receipts = set((mini_repo / "operations/transactions").glob("transaction-*.yaml"))
+    # The fixture has no remote: commit is real and push fails without network access.
+    args = ["--commit-message", "fixture partial completion", "--push"]
+    step = ({"id": "end", "use": "session-end", "args": args} if use else
+            {"id": "end", "argv": ["session-end", *args]})
+    report = agent_run.run_plan({"session_id": session, "on_error": "continue", "steps": [
+        step, {"id": "later", "kind": "log", "text": "must not run", "key": "after-partial-porcelain"},
+        {"id": "read", "argv": ["inspect", "note-demo"]},
+    ]}, mini_repo)
+    assert subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=mini_repo, text=True) != before
+    assert session_store.load_session_ledger(mini_repo, session)["state"] == "committed_unpushed"
+    assert set((mini_repo / "operations/transactions").glob("transaction-*.yaml")) == receipts
+    assert report["steps"][0]["status"] == "error", report
+    assert report["steps"][0]["commit_certainty"] == "uncertain"
+    assert report["steps"][1]["status"] == "skipped"
+    assert report["steps"][1]["error"]["code"] == "PLAN_STOPPED"
+    assert report["steps"][2]["status"] == "ok"
+
+
+@pytest.mark.parametrize("response", ["exception", "error-with-none", "success-without-evidence"])
+def test_porcelain_without_commit_evidence_retains_uncertainty(mini_repo, monkeypatch, response):
+    from types import SimpleNamespace
+
+    from learning_os.adapters import cli
+
+    calls = []
+
+    def dispatch(namespace):
+        calls.append(namespace.command)
+        if namespace.command == "session-end":
+            if response == "exception":
+                raise RuntimeError("fixture transport failed after dispatch")
+            if response == "error-with-none":
+                return SimpleNamespace(exit_code=1, stdout='{"ok":false,"commit":"none"}', stderr="fixture failed")
+        return SimpleNamespace(exit_code=0, stdout='{"ok":true}', stderr="")
+
+    monkeypatch.setattr(cli, "invoke_outcome", dispatch)
+    report = agent_run.run_plan({"session_id": "porcelain-evidence", "on_error": "continue", "steps": [
+        {"id": "end", "argv": ["session-end"]},
+        {"id": "later", "argv": ["generate"]},
+        {"id": "read", "argv": ["inspect", "note-demo"]},
+    ]}, mini_repo)
+    assert calls == ["session-end", "inspect"], report
+    assert report["steps"][0]["commit_certainty"] == "uncertain"
+    assert report["steps"][1]["status"] == "skipped"
+    assert report["steps"][2]["status"] == "ok"
+
+
+@pytest.mark.parametrize("receipt", [False, True])
+def test_gateway_refusal_cannot_erase_commit_receipt(mini_repo, monkeypatch, receipt):
+    payload = {"ok": False, "commit": "none"}
+    if receipt:
+        payload.update(_receipt(mini_repo))
+        payload["ok"] = False
+    monkeypatch.setattr(agent_run.subprocess, "run", lambda argv, **kwargs:
+                        subprocess.CompletedProcess(argv, 1, json.dumps(payload), "fixture refusal"))
+    report = agent_run.run_plan({"session_id": "gateway-evidence", "steps": [
+        {"id": "write", "kind": "log", "text": "fixture", "key": "gateway-evidence"},
+    ]}, mini_repo)
+    assert report["steps"][0]["status"] == "error"
+    assert report["steps"][0]["commit_certainty"] == ("committed" if receipt else "none")
+
+
 def test_external_feed_reads_are_neither_memoized_nor_parallel(mini_repo, tmp_path):
     feed = tmp_path / 'external-feed.json'
     feed.write_text('{}')

tests/test_projection_readiness.py

@@ -6,6 +6,8 @@
 import sys
 from pathlib import Path
 
+import pytest
+
 from learning_os.app.projection_readiness import (
     dependency_ready,
     observe,
@@ -64,3 +66,26 @@ def test_forged_receipt_is_not_completion(mini_repo):
     path.parent.mkdir(parents=True, exist_ok=True)
     path.write_text("status: committed\nsnapshot_after: sha256:fake\n")
     assert not receipt_completed(mini_repo, str(path.relative_to(mini_repo)), "sha256:fake")
+
+
+@pytest.mark.parametrize("relative", [
+    "knowledge/notes/fake.yaml", "operations/transactions/../../knowledge/fake.yaml",
+    "operations/transactions-other/fake.yaml", "../outside.yaml",
+])
+def test_receipt_paths_are_rejected_before_authority_read(mini_repo, monkeypatch, relative):
+    from learning_os.app import projection_readiness
+
+    def unexpected(*args):
+        pytest.fail("receipt outside the transaction boundary was opened")
+
+    monkeypatch.setattr(projection_readiness, "_completed", unexpected)
+    assert not receipt_completed(mini_repo, relative, "sha256:fake")
+
+
+def test_receipt_symlink_cannot_escape_transaction_boundary(mini_repo):
+    target = mini_repo / "knowledge/fake-receipt.yaml"
+    target.write_text("status: committed\nsnapshot_after: sha256:fake\n")
+    path = mini_repo / "operations/transactions/linked.yaml"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.symlink_to(target)
+    assert not receipt_completed(mini_repo, str(path.relative_to(mini_repo)), "sha256:fake")

tools/README.md

@@ -51,7 +51,7 @@ runtime coverage; the exact roots and historical allowlist live in
 | `diagnostics_prune.py` | bounded maintenance | Prunes the disposable diagnostics trace store to its retention bound. |
 | `affected_tests.py` | supported test tool | Maps changed files to test groups for `make test-affected`; a rerun minimizer, never a gate. |
 | `seal_envelope.py` | bounded authoring tool | Seals one GatewayEnvelopeV2 from caller intent (WORKFLOWS §25c); reads the repo, writes nothing canonical. |
-| `agent_run.py` | experimental operator runner | One call batches CLI steps over an explicitly selected data root. Log/gateway plans require a stable `session_id`; logs also require unique explicit retry keys. Presealed gateway files must bind the same session, and approved bytes are preserved. `on_error` defaults to `stop`, reporting skipped steps; `continue` keeps independent steps running. Dependent reads require a projection from a successful write at or after the latest required write. Missing/pending readiness and unknown snapshots refuse. Nonfinite output fails its step; summaries name `_missing_fields` under their UTF-8 byte budget. Optional typed preflight reuses the pinned validation cache and refuses every validation error. Cluster previews retain counts and a supported expansion. Bounded read sessions memoize eligible reads under manifest pins; `read_workers` (1-4, default 1) runs contiguous eligible read waves in isolated processes; `deadline_s` (default 120) stops admission; `report_max_bytes` bounds the whole report with a private full-report artifact on every run (`--report-dir` chooses its parent). `--expand-report PATH [--step ID]` verifies report and raw-output digests before expansion. No bulk learner decisions. |
+| `agent_run.py` | experimental operator runner | One call batches CLI steps over an explicitly selected data root. Plans with derived, canonical or undeclared effects require a stable `session_id`; only operations explicitly declaring read effects count as reads, consistently for `argv` and `use`; logs also require unique explicit retry keys. Presealed gateway files must bind the same session, and approved bytes are preserved. `on_error` defaults to `stop`, reporting skipped steps; `continue` keeps independent steps running. Dependent reads require a projection from a successful write at or after the latest required write. Missing/pending readiness and unknown snapshots refuse. Nonfinite output fails its step; summaries name `_missing_fields` under their UTF-8 byte budget. Optional typed preflight reuses the pinned validation cache and refuses every validation error. Cluster previews retain counts and a supported expansion. Bounded read sessions memoize eligible reads under manifest pins; `read_workers` (1-4, default 1) runs contiguous eligible read waves in isolated processes; `deadline_s` (default 120) stops admission; `report_max_bytes` bounds the whole report with a private full-report artifact on every run (`--report-dir` chooses its parent). `--expand-report PATH [--step ID]` verifies report and raw-output digests before expansion. No bulk learner decisions. |
 | `capture_world.py` | supported contract tool | Captures one world (data, pins, observations, material inventory) for captured-world identity (SPEC §6). |
 | `compare_world.py` | supported gate | Compares two revisions against one captured world; exits 0 on identity, 1 on difference (SPEC §6). |