fix(task): standardize blocked precondition receipts

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+### Fixed
+
+- Emit deterministic blocked-precondition schemas and stderr diagnostics for dry-run task receipts (#122).
+
 ## [0.18.6] - 2026-08-02
 
 ### Changed

docs/blocked-preconditions.md

@@ -1,15 +1,26 @@
 # Structured blocked-precondition receipts
 
-`simplicio-py task --dry-run-task --json` is fail-closed when required preconditions are missing. It exits non-zero and emits JSON with `status: blocked`, `applied: false`, and `blocked_preconditions`. Each entry contains stable `reason`, actionable `next_surface`, and a human-readable `message`.
+`simplicio-py task --dry-run-task --json` is fail-closed when required preconditions are missing. It exits non-zero and emits exactly one `simplicio.dev-cli.task-result/v1` JSON object on stdout. Human diagnostics remain on stderr.
+
+A blocked task result always includes `status: "blocked"`, `applied: false`, `model_invoked`, `next_surface`, `reason_code`, `execution_profile`, and `blocked_preconditions`. Each precondition uses the deterministic `simplicio.dev-cli.blocked-precondition/v1` shape:
 
 ```json
-{"status":"blocked","applied":false,"blocked_preconditions":[{"reason":"artifacts_missing","next_surface":"mapper_artifacts","message":"mapper artifacts are required before task generation"}]}
+{
+  "schema": "simplicio.dev-cli.blocked-precondition/v1",
+  "code": "artifacts_missing",
+  "reason": "artifacts_missing",
+  "message": "mapper artifacts required for dry-run task are missing",
+  "next_surface": "mapper_artifacts",
+  "next_action": "generate fresh Mapper artifacts, then retry",
+  "retryable": true,
+  "details": {"missing": ["project_map", "precedent_index"]}
+}
 ```
 
-Route `mapper_artifacts` to the mapper, `mapper_inspection` to inspect or await its job, and `context_pack` to provide missing handoff context. Dry-run never applies a patch or runs tests. Treat unknown reasons as blocked and preserve the complete receipt for evidence.
+Route `mapper_artifacts` to Mapper generation, `mapper_inspection` to stale-artifact inspection, `context_pack` to a fresh handoff/context pack, and `task_target` to target selection. Provider and execution-mode blocks name those surfaces directly. Treat unknown reasons as blocked and preserve the complete receipt for evidence.
 
 ```bash
 simplicio-py task "implement the change" --stack python --target src/module.py --dry-run-task --json > receipt.json
 ```
 
-The schema is additive; preserve `blocked_preconditions`, `warnings`, and `next_surface` for operators.
+The JSON schema is additive. Consumers must preserve unknown top-level fields, retain every `blocked_preconditions` entry, and use `reason`/`code` plus `next_surface` instead of parsing human messages. Diagnostics such as `BLOCKED[artifacts_missing]: ...; next_surface=mapper_artifacts` are stderr-only and never corrupt stdout JSON.

simplicio/commands/task.py

@@ -16,6 +16,18 @@
 from ._shared import force_local_if_requested
 
 
+def _emit_blocked_diagnostics(result: dict) -> None:
+    if result.get("status") != "blocked":
+        return
+    for blocker in result.get("blocked_preconditions", []):
+        if not isinstance(blocker, dict):
+            continue
+        code = blocker.get("code") or blocker.get("reason") or "blocked_precondition"
+        message = blocker.get("message") or code
+        next_surface = blocker.get("next_surface") or "task_preconditions"
+        print(f"BLOCKED[{code}]: {message}; next_surface={next_surface}", file=sys.stderr)
+
+
 def _run_verification_only(a: argparse.Namespace) -> int:
     from ..pipeline_stages import _configured_test_command, _verification_timeout_seconds
     from ..runtime_env import prepare_project_command
@@ -176,9 +188,7 @@ def run(a: argparse.Namespace) -> int:
         )
         if a.json:
             print(json.dumps(result, sort_keys=True))
-            terminal = result.get("provider_terminal")
-            if isinstance(terminal, dict) and terminal.get("message"):
-                print(str(terminal["message"]), file=sys.stderr)
+            _emit_blocked_diagnostics(result)
         else:
             status = (
                 "BLOCKED" if result.get("status") == "blocked" else ("DRY-RUN" if a.dry_run_task else "DONE")

simplicio/pipeline.py

@@ -386,7 +386,7 @@ def _run_dry_run_task(
         allow_degraded_mapper=requested_execution_mode == "standalone",
     )
     if blockers:
-        return _task_result(
+        result = _task_result(
             target,
             prompt,
             "",
@@ -396,6 +396,12 @@ def _run_dry_run_task(
             blocked_preconditions=blockers,
             target_kind=target_kind(root, target),
         )
+        result["execution_profile"] = {
+            "requested_mode": str(requested_execution_mode or "auto"),
+            "effective_mode": "blocked",
+            "reason_code": str(blockers[0]["reason"]),
+        }
+        return result
     if os.environ.get("SIMPLICIO_STANDALONE_PREFLIGHT", "").strip().lower() in {
         "1",
         "true",

simplicio/pipeline_task_result.py

@@ -28,6 +28,9 @@
 from .prompt import latest_prompt_envelope
 from .providers import _provider_id
 
+TASK_RESULT_SCHEMA = "simplicio.dev-cli.task-result/v1"
+BLOCKED_PRECONDITION_SCHEMA = "simplicio.dev-cli.blocked-precondition/v1"
+
 
 def _verify_receipt_payload(receipt: dict[str, Any] | None) -> dict[str, Any] | None:
     if not receipt:
@@ -56,6 +59,48 @@ def _diff_summary(files_changed):
     return "changed " + ", ".join(files_changed)
 
 
+_NEXT_SURFACE_BY_REASON = {
+    "CONTEXT_REQUIRED": "context_pack",
+    "MAPPER_CONTEXT_IDENTITY_REQUIRED": "context_pack",
+    "SOURCE_DRIFT": "context_snapshot",
+    "TARGET_OUTSIDE_SCOPE": "task_target",
+    "TASK_SPEC_REQUIRES_INTEGRATED_MODE": "execution_mode",
+    "plan_compilation_failed": "task_plan",
+    "target_outside_root": "task_target",
+}
+_NEXT_ACTION_BY_SURFACE = {
+    "context_pack": "provide a fresh Mapper context pack, then retry",
+    "context_snapshot": "refresh the Mapper context snapshot, then retry",
+    "execution_mode": "select the required execution mode, then retry",
+    "mapper_artifacts": "generate fresh Mapper artifacts, then retry",
+    "mapper_inspection": "refresh stale Mapper artifacts, then retry",
+    "provider": "resolve the provider precondition, then retry",
+    "task_plan": "correct the task plan, then retry",
+    "task_target": "select a valid task target, then retry",
+}
+
+
+def _normalize_blocked_precondition(value: dict[str, Any]) -> dict[str, Any]:
+    reason = str(value.get("reason") or value.get("code") or "blocked_precondition")
+    code = str(value.get("code") or reason)
+    next_surface = str(value.get("next_surface") or _NEXT_SURFACE_BY_REASON.get(reason, "task_preconditions"))
+    next_action = str(
+        value.get("next_action")
+        or _NEXT_ACTION_BY_SURFACE.get(next_surface, "resolve the blocked precondition, then retry")
+    )
+    details = value.get("details")
+    return {
+        "schema": BLOCKED_PRECONDITION_SCHEMA,
+        "code": code,
+        "reason": reason,
+        "message": str(value.get("message") or reason),
+        "next_surface": next_surface,
+        "next_action": next_action,
+        "retryable": bool(value.get("retryable", True)),
+        "details": dict(details) if isinstance(details, dict) else {},
+    }
+
+
 def _degraded_mapper_context_allowed(context_pack: dict[str, Any] | None) -> bool:
     """Allow only Loop-issued, explicit degraded context in standalone mode."""
     if not isinstance(context_pack, dict):
@@ -253,6 +298,7 @@ def _task_result(
     model = os.environ.get("SIMPLICIO_MODEL", "")
     cost_usd = float(_estimate_price(model, prompt_tokens, completion_tokens)) if priced else 0.0
     result = {
+        "schema": TASK_RESULT_SCHEMA,
         "task_id": task_id,
         "applied": bool(applied),
         "status": status or ("applied" if applied else "failed"),
@@ -281,7 +327,14 @@ def _task_result(
     if envelope is not None:
         result["prompt_envelope"] = envelope.receipt()
     if blocked_preconditions:
-        result["blocked_preconditions"] = blocked_preconditions
+        normalized = [
+            _normalize_blocked_precondition(item) for item in blocked_preconditions if isinstance(item, dict)
+        ]
+        if normalized:
+            result["blocked_preconditions"] = normalized
+            result["model_invoked"] = bool(output)
+            result["next_surface"] = normalized[0]["next_surface"]
+            result["reason_code"] = normalized[0]["code"]
     verify_receipt = _verify_receipt_payload(verify)
     if verify_receipt is not None:
         exit_codes = verify_receipt.get("exit_codes", [])

tests/python/test_pipeline_task_result_module_coverage_gap.py

@@ -307,7 +307,18 @@ def test_task_result_blocked_preconditions_included(monkeypatch):
     result = ptr._task_result(
         "T01", "prompt", "", applied=False, status="blocked", blocked_preconditions=blockers
     )
-    assert result["blocked_preconditions"] == blockers
+    assert result["blocked_preconditions"] == [
+        {
+            "schema": "simplicio.dev-cli.blocked-precondition/v1",
+            "code": "x",
+            "reason": "x",
+            "message": "m",
+            "next_surface": "task_preconditions",
+            "next_action": "resolve the blocked precondition, then retry",
+            "retryable": True,
+            "details": {},
+        }
+    ]
 
 
 def test_task_result_prompt_envelope_receipt(monkeypatch):

tests/python/test_task_json_contract.py

@@ -1,5 +1,8 @@
 import json
+import os
+import subprocess
 import sys
+from pathlib import Path
 
 from simplicio import cli
 
@@ -273,16 +276,88 @@ def fail_if_called(*_args, **_kwargs):
         ]
     )
 
+    captured = capsys.readouterr()
     assert code == 1
-    payload = json.loads(capsys.readouterr().out)
+    payload = json.loads(captured.out)
+    assert payload["schema"] == "simplicio.dev-cli.task-result/v1"
     assert payload["status"] == "blocked"
-    assert payload["blocked_preconditions"][0]["next_surface"]
+    assert payload["applied"] is False
+    assert payload["model_invoked"] is False
+    assert payload["next_surface"] == "mapper_artifacts"
+    assert payload["execution_profile"] == {
+        "requested_mode": "auto",
+        "effective_mode": "blocked",
+        "reason_code": "artifacts_missing",
+    }
+    expected_keys = {
+        "schema",
+        "code",
+        "reason",
+        "message",
+        "next_surface",
+        "next_action",
+        "retryable",
+        "details",
+    }
+    assert all(set(item) == expected_keys for item in payload["blocked_preconditions"])
     reasons = {item["reason"] for item in payload["blocked_preconditions"]}
     assert "artifacts_missing" in reasons
     assert "no_handoff_targets" in reasons
+    assert "BLOCKED[artifacts_missing]" in captured.err
     assert called["generate"] == 0
 
 
+def test_task_dry_run_subprocess_emits_single_actionable_json_receipt(tmp_path):
+    _write(tmp_path / "app.py", "old\n")
+    repo_root = Path(__file__).resolve().parents[2]
+    env = os.environ.copy()
+    env.update(
+        {
+            "PYTHONPATH": str(repo_root),
+            "SIMPLICIO_MAPPER_CLI": "0",
+            "SIMPLICIO_SKIP_AUTO_INIT": "1",
+        }
+    )
+
+    completed = subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "simplicio.cli",
+            "task",
+            "update app",
+            "--root",
+            str(tmp_path),
+            "--target",
+            "app.py",
+            "--dry-run-task",
+            "--json",
+        ],
+        cwd=repo_root,
+        env=env,
+        capture_output=True,
+        text=True,
+        timeout=30,
+        check=False,
+    )
+
+    payload = json.loads(completed.stdout)
+    assert completed.returncode == 1
+    assert completed.stdout.count("\n") == 1
+    assert payload["schema"] == "simplicio.dev-cli.task-result/v1"
+    assert payload["status"] == "blocked"
+    assert payload["applied"] is False
+    assert payload["model_invoked"] is False
+    assert payload["next_surface"] == "mapper_artifacts"
+    assert payload["execution_profile"]["effective_mode"] == "blocked"
+    assert {item["reason"] for item in payload["blocked_preconditions"]} == {
+        "artifacts_missing",
+        "no_handoff_targets",
+    }
+    assert "BLOCKED[artifacts_missing]" in completed.stderr
+    assert "BLOCKED[no_handoff_targets]" in completed.stderr
+
+
 def test_task_dry_run_json_serializes_provider_block_with_context_reason(tmp_path, monkeypatch, capsys):
     from simplicio.providers import ProviderExecutionError