diff --git a/loop/contract.py b/loop/contract.py index 960643b..9cbf821 100644 --- a/loop/contract.py +++ b/loop/contract.py @@ -59,6 +59,25 @@ def _parse_scalar(value: str) -> Any: return value +def _strip_comment(raw: str) -> str: + """Drop a trailing ``#`` comment, but only when the ``#`` is unquoted. + + A ``#`` inside a single- or double-quoted scalar is data (``"reach #1"``), + not a comment; and — like YAML — a ``#`` only opens a comment at line start + or after whitespace, so an unquoted ``reach#1`` is left intact. + """ + + in_single = in_double = False + for i, ch in enumerate(raw): + if ch == "'" and not in_double: + in_single = not in_single + elif ch == '"' and not in_single: + in_double = not in_double + elif ch == "#" and not in_single and not in_double and (i == 0 or raw[i - 1] in " \t"): + return raw[:i] + return raw + + def _fallback_yaml(text: str) -> dict[str, Any]: """Small YAML subset parser for the manifest shapes this project emits. @@ -69,7 +88,7 @@ def _fallback_yaml(text: str) -> dict[str, Any]: root: dict[str, Any] = {} current_key: str | None = None for raw in text.splitlines(): - line = raw.split("#", 1)[0].rstrip() + line = _strip_comment(raw).rstrip() if not line: continue if not line.startswith(" ") and ":" in line: @@ -187,8 +206,10 @@ def _check_terminal_contradiction(data: dict[str, Any] | None, path: Path, issue """G1: a Succeeded terminal must not contradict its own evidence. Runs in both validation modes because JSON Schema cannot express the - cross-field rule that a success claim requires false_completion=false AND at - least one met criterion. + cross-field rule that a success claim requires false_completion=false, at + least one met criterion, AND a non-empty evidence list — mirroring the + write-time refusal in loop/emit.py (an evidence-free Succeeded is a claim + with nothing behind it). """ if not isinstance(data, dict) or data.get("state") != "Succeeded": @@ -202,6 +223,11 @@ def _check_terminal_contradiction(data: dict[str, Any] | None, path: Path, issue issues.append( ContractIssue("contradictory_terminal", "Succeeded terminal has no met (true) entry in criteria_met", path) ) + evidence = data.get("evidence") + if not isinstance(evidence, list) or not evidence: + issues.append( + ContractIssue("contradictory_terminal", "Succeeded terminal has empty evidence[] (G1)", path) + ) def _validate_terminal(data: dict[str, Any] | None, path: Path, issues: list[dict]) -> None: @@ -246,7 +272,67 @@ def _verify_script_paths(paths: LoopPaths) -> list[Path]: return [scripts / "verify-fast", scripts / "verify-fast.sh", scripts / "verify-full", scripts / "verify-full.sh"] +def _task_verify_values(tasks: dict[str, Any] | None) -> list[str]: + """Non-empty ``verify`` strings declared by the tasks, in order.""" + if not isinstance(tasks, dict): + return [] + task_list = tasks.get("tasks") + if not isinstance(task_list, list): + return [] + values: list[str] = [] + for task in task_list: + if isinstance(task, dict): + verify = task.get("verify") + if isinstance(verify, str) and verify.strip(): + values.append(verify.strip()) + return values + + +def _check_verify_surface(paths: LoopPaths, tasks: dict[str, Any] | None, issues: list[dict]) -> None: + """Every loop needs a verification surface, and a declared one must resolve. + + (a) If NO verify-* script exists AND no task declares a verify command, the + contract can never gate anything — ``missing_verify_surface``. + (b) A task ``verify`` whose first whitespace token is path-shaped (contains + "/") must resolve relative to the workspace — a dangling script path is + ``unresolved_task_verify``. A plain command (``pytest -q``) is not a path + and is not existence-checked. + + Runs in both validation modes. + """ + any_script = any(p.is_file() for p in _verify_script_paths(paths)) + verify_values = _task_verify_values(tasks) + if not any_script and not verify_values: + issues.append( + ContractIssue( + "missing_verify_surface", + "no verify-* script exists and no task declares a verify command", + paths.tasks, + ) + ) + for value in verify_values: + first = value.split()[0] + if "/" not in first: + continue + if not (paths.workspace / first).exists(): + issues.append( + ContractIssue( + "unresolved_task_verify", + f"task verify path {first!r} does not resolve under the workspace", + paths.tasks, + ) + ) + + def _check_stub_verify_scripts(paths: LoopPaths, issues: list[dict]) -> None: + """Flag verify-* scripts that still carry the scaffold's stub markers. + + The ``stub:`` / ``replace with real command`` markers are an OPT-IN + convention: a loop that wants doctor to refuse an un-filled gate leaves them + in until the real command lands. The templates ship WITHOUT them so a fresh + scaffold is doctor-clean — the presence of a marker is a deliberate signal, + never the default state. + """ for script in _verify_script_paths(paths): if not script.exists() or not script.is_file(): continue @@ -344,8 +430,22 @@ def _validate_record(data: dict[str, Any], schema_key: str, path: Path, mode: st _structural_record_check(data, _load_schema_file(filename), path, issues) +# The canonical rollout / candidate ledger file (scripts/rollout_ledger.py, +# schemas/rollout-record.schema.json). doctor validates THIS as a rollout ledger; +# any other .loop/*.jsonl is foreign and skipped rather than force-validated +# against the rollout schema. +ROLLOUT_LEDGER_NAMES = ("rollout.jsonl",) + + def _validate_jsonl(path: Path, schema_key: str, mode: str, issues: list[dict]) -> None: - for lineno, line in enumerate(path.read_text(encoding="utf-8", errors="ignore").splitlines(), start=1): + try: + text = path.read_text(encoding="utf-8") + except UnicodeDecodeError as exc: + # A ledger that is not valid UTF-8 fails closed — decoding it lossily + # (errors="ignore") would let a record with corrupt bytes validate clean. + issues.append(ContractIssue("invalid_encoding", f"{path.name}: not valid UTF-8: {exc}", path)) + return + for lineno, line in enumerate(text.splitlines(), start=1): line = line.strip() if not line: continue @@ -372,9 +472,11 @@ def _validate_optional_records(paths: LoopPaths, mode: str, issues: list[dict]) if data is not None: _validate_record(data, "repair", record_path, mode, issues) checked.add("repair") - for ledger_path in sorted(paths.loop_dir.glob("*.jsonl")): - _validate_jsonl(ledger_path, "rollout", mode, issues) - checked.add("rollout") + for name in ROLLOUT_LEDGER_NAMES: + ledger_path = paths.loop_dir / name + if ledger_path.is_file(): + _validate_jsonl(ledger_path, "rollout", mode, issues) + checked.add("rollout") receipts_dir = paths.loop_dir / "receipts" if receipts_dir.is_dir(): for receipt_path in sorted(receipts_dir.glob("*.jsonl")): @@ -421,6 +523,7 @@ def validate_contract(target: str | Path) -> dict[str, Any]: if not paths.runlog.exists(): issues.append(ContractIssue("missing_file", "missing RUNLOG.md", paths.runlog)) _check_stub_verify_scripts(paths, issues) + _check_verify_surface(paths, tasks, issues) records_checked = _validate_optional_records(paths, mode, issues) schemas_checked = list(SCHEMA_IDS) + [ diff --git a/loop/paths.py b/loop/paths.py index e9f2d61..1b03689 100644 --- a/loop/paths.py +++ b/loop/paths.py @@ -25,6 +25,12 @@ def to_json(self) -> dict[str, str]: def _workspace_from(target: Path) -> Path: target = target.resolve() + if target.is_file(): + # A FILE target (.loop/state.json, TASKS.json, RUNLOG.md, …) names a + # contract artifact, not the workspace. Resolve from its parent so the + # owning workspace is found instead of treating the file as a directory + # and building garbage paths underneath it. + target = target.parent if target.name == ".loop": return target.parent if (target / ".loop").is_dir(): diff --git a/scripts/test_contract_records.py b/scripts/test_contract_records.py index 92d6f9b..3d166e8 100644 --- a/scripts/test_contract_records.py +++ b/scripts/test_contract_records.py @@ -85,6 +85,55 @@ def test_present_rollout_jsonl_bad_line_is_flagged(tmp_path): assert any("rollout.jsonl" in i["message"] for i in issues) +def test_foreign_jsonl_is_not_validated_as_rollout(tmp_path): + # F5a: doctor must validate only the canonical rollout ledger (rollout.jsonl), + # not every .loop/*.jsonl. A foreign notes.jsonl must not false-FAIL a healthy + # contract by being force-validated against the rollout schema. + loop_dir = tmp_path / ".loop" + loop_dir.mkdir() + (loop_dir / "notes.jsonl").write_text(json.dumps({"note": "hi"}) + "\n", encoding="utf-8") + assert _optional_issues(tmp_path) == [] + + +def test_foreign_jsonl_does_not_mark_rollout_checked(tmp_path): + # F5a: "rollout" belongs in schemas_checked only when a canonical ledger was + # actually validated — an unknown jsonl must not inflate coverage. + loop_dir = tmp_path / ".loop" + loop_dir.mkdir() + (loop_dir / "notes.jsonl").write_text(json.dumps({"note": "hi"}) + "\n", encoding="utf-8") + checked = _validate_optional_records(resolve_loop_paths(tmp_path), "structural-fallback", []) + assert "rollout" not in checked + + +def test_canonical_rollout_ledger_is_still_validated_and_marked(tmp_path): + # F5a: the canonical rollout.jsonl must still be validated and reported. + loop_dir = tmp_path / ".loop" + loop_dir.mkdir() + good = {"id": "c1", "parent": None, "verdict": "ok", "score": 0.9, + "score_delta": 0.1, "coherent_with_prior_winner": True, "productive": True} + (loop_dir / "rollout.jsonl").write_text(json.dumps(good) + "\n", encoding="utf-8") + issues: list[dict] = [] + checked = _validate_optional_records(resolve_loop_paths(tmp_path), "structural-fallback", issues) + assert issues == [] + assert "rollout" in checked + + +def test_rollout_ledger_with_invalid_utf8_is_flagged(tmp_path): + # F5b: a rollout ledger line carrying a raw 0xff byte inside an otherwise-valid + # JSON record must not silently validate clean under errors="ignore" (a false + # PASS). Strict decode fails the file closed with an invalid_encoding issue. + loop_dir = tmp_path / ".loop" + loop_dir.mkdir() + prefix = b'{"id": "c1", "parent": null, "verdict": "o' + suffix = ( + b'k", "score": 0.9, "score_delta": 0.1, ' + b'"coherent_with_prior_winner": true, "productive": true}' + ) + (loop_dir / "rollout.jsonl").write_bytes(prefix + b"\xff" + suffix + b"\n") + issues = _optional_issues(tmp_path) + assert any(i["code"] == "invalid_encoding" for i in issues), issues + + def test_present_valid_receipt_jsonl_passes(tmp_path): receipts = tmp_path / ".loop" / "receipts" receipts.mkdir(parents=True) diff --git a/scripts/test_loop_contract_core.py b/scripts/test_loop_contract_core.py index b49dfe6..3acf9dd 100644 --- a/scripts/test_loop_contract_core.py +++ b/scripts/test_loop_contract_core.py @@ -190,6 +190,41 @@ def test_read_manifest_returns_dict_on_malformed_yaml(tmp_path): assert isinstance(result, dict) +def test_f6_fallback_yaml_keeps_hash_inside_quotes(): + # F6: the fallback parser stripped everything after the first '#' before it + # ever looked at quotes, so goal: "reach #1" truncated to '"reach'. A '#' + # inside a quoted scalar is data, not a comment. + from loop.contract import _fallback_yaml + + assert _fallback_yaml('goal: "reach #1"\n') == {"goal": "reach #1"} + assert _fallback_yaml("goal: 'reach #1'\n") == {"goal": "reach #1"} + # A genuine unquoted trailing comment is still stripped. + assert _fallback_yaml("goal: ship # real comment\n") == {"goal": "ship"} + # And a nested (one-level) mapping value keeps its quoted '#' too. + assert _fallback_yaml('policies:\n note: "see #3"\n') == {"policies": {"note": "see #3"}} + + +def test_f6_both_manifest_parse_paths_agree_on_quoted_hash(tmp_path, monkeypatch): + # F6: PyYAML and the fallback subset parser must agree that a '#' inside a + # quoted value survives. Pin both paths — the real library, and the fallback + # forced by hiding the yaml module (import yaml -> ImportError -> fallback). + import sys + + from loop.contract import read_manifest + + manifest = tmp_path / "manifest.yaml" + manifest.write_text('schema: loop-engineer/manifest@1\ngoal: "reach #1"\n', encoding="utf-8") + + yaml = pytest.importorskip("yaml") + real = read_manifest(manifest) + assert real["goal"] == "reach #1" + + monkeypatch.setitem(sys.modules, "yaml", None) # import yaml now raises ImportError + fallback = read_manifest(manifest) + assert fallback["goal"] == "reach #1" + assert fallback["goal"] == real["goal"] + + def test_doctor_does_not_crash_on_malformed_manifest(tmp_path): # F1 (blast radius): the doctor CLI validates a foreign contract via the same # read_manifest; a malformed manifest must produce an actionable report, not a @@ -251,12 +286,87 @@ def test_g1_contradictory_succeeded_terminal_emits_issue(): assert any(i["code"] == "contradictory_terminal" for i in all_false_issues) happy_issues = _terminal_issues( - {**base, "criteria_met": {"1": True}, "false_completion": False} + {**base, "criteria_met": {"1": True}, "false_completion": False, "evidence": ["e.json"]} ) assert not any(i["code"] == "contradictory_terminal" for i in happy_issues) assert happy_issues == [] +def _write_valid_succeeded_contract(root: pathlib.Path, evidence) -> pathlib.Path: + """A doctor-clean scaffold mutated into a Succeeded terminal with the given + evidence list. Used to pin F1 end-to-end (both validation modes).""" + from loop.scaffold import scaffold + + target = root / "loop" + scaffold(target) + state_path = target / ".loop" / "state.json" + state = json.loads(state_path.read_text(encoding="utf-8")) + state["terminal_state"] = "Succeeded" + state_path.write_text(json.dumps(state), encoding="utf-8") + (target / ".loop" / "terminal_state.json").write_text( + json.dumps( + { + "schema": "loop-engineer/terminal@1", + "project": "loop", + "state": "Succeeded", + "iteration_id": 1, + "criteria_met": {"c1": True}, + "evidence": evidence, + "false_completion": False, + "terminated_at": "2026-01-01T00:00:00+00:00", + } + ), + encoding="utf-8", + ) + return target + + +def test_g1_empty_evidence_succeeded_terminal_is_flagged_unit(): + # F1: a Succeeded terminal with no evidence outruns its own claim exactly like + # false_completion=true or an unmet criterion. _validate_terminal runs the + # cross-field check in BOTH validation modes, so this unit assertion covers both. + empty_evidence_issues = _terminal_issues( + { + "schema": "loop-engineer/terminal@1", + "state": "Succeeded", + "criteria_met": {"c1": True}, + "false_completion": False, + "evidence": [], + } + ) + assert any(i["code"] == "contradictory_terminal" for i in empty_evidence_issues) + assert any( + "evidence" in i["message"] + for i in empty_evidence_issues + if i["code"] == "contradictory_terminal" + ) + + +def test_f1_empty_evidence_succeeded_fails_doctor_end_to_end(tmp_path): + # F1 repro: a schema-valid contract whose terminal declares Succeeded with an + # empty evidence[] must NOT pass doctor. Runs under whichever validation mode + # is installed; the two suite invocations exercise both. + from loop.contract import doctor_report + + target = _write_valid_succeeded_contract(tmp_path, evidence=[]) + report = doctor_report(target) + assert report["ok"] is False, report["issues"] + assert any( + i["code"] == "contradictory_terminal" and "evidence" in i["message"] + for i in report["issues"] + ) + + +def test_f1_non_empty_evidence_succeeded_still_passes_doctor(tmp_path): + # The fix must not over-fire: a Succeeded terminal that DOES carry evidence + # stays doctor-clean. + from loop.contract import doctor_report + + target = _write_valid_succeeded_contract(tmp_path, evidence=[".loop/artifacts/verify-T1.json"]) + report = doctor_report(target) + assert report["ok"] is True, report["issues"] + + def test_jsonschema_mode_rejects_schema_violating_artifact(tmp_path): # M1-SCHEMAS: when jsonschema is installed the doctor must run REAL schema # validation, not just the weaker structural hand checks — a state.json @@ -328,6 +438,121 @@ def test_jsonschema_mode_enforces_every_schema_required_field(tmp_path): assert any(i["code"] == "schema_violation" for i in report["issues"]), field +def _scaffold(tmp_path: pathlib.Path, name: str) -> pathlib.Path: + from loop.scaffold import scaffold + + target = tmp_path / name + scaffold(target) + return target + + +def _set_task_verify(target: pathlib.Path, value) -> None: + tasks_path = target / "TASKS.json" + tasks = json.loads(tasks_path.read_text(encoding="utf-8")) + tasks["tasks"][0]["verify"] = value + tasks_path.write_text(json.dumps(tasks), encoding="utf-8") + + +def test_f2_no_verify_surface_is_flagged(tmp_path): + # F2(a): a contract with no verify-* script AND no task declaring a verify + # command has no verification surface at all — doctor must say so. + from loop.contract import doctor_report + + target = _scaffold(tmp_path, "no-surface") + (target / "scripts" / "verify-fast").unlink() + (target / "scripts" / "verify-full").unlink() + _set_task_verify(target, "") + + report = doctor_report(target) + assert report["ok"] is False + assert any(i["code"] == "missing_verify_surface" for i in report["issues"]), report["issues"] + + +def test_f2_scaffold_with_deleted_verify_scripts_is_flagged(tmp_path): + # F2(b) repro: deleting scripts/verify-* from a scaffold left doctor green + # even though every task still points at the now-missing script. + from loop.contract import doctor_report + + target = _scaffold(tmp_path, "deleted-scripts") + (target / "scripts" / "verify-fast").unlink() + (target / "scripts" / "verify-full").unlink() + + report = doctor_report(target) + assert report["ok"] is False + assert any(i["code"] == "unresolved_task_verify" for i in report["issues"]), report["issues"] + + +def test_f2_unresolvable_path_shaped_task_verify_is_flagged(tmp_path): + # F2(b): a path-shaped task.verify that does not resolve relative to the + # workspace is flagged, even when the verify-* scripts exist. + from loop.contract import doctor_report + + target = _scaffold(tmp_path, "bad-path") + _set_task_verify(target, "scripts/does-not-exist") + + report = doctor_report(target) + assert report["ok"] is False + assert any(i["code"] == "unresolved_task_verify" for i in report["issues"]), report["issues"] + assert not any(i["code"] == "missing_verify_surface" for i in report["issues"]) + + +def test_f2_plain_command_task_verify_is_not_path_checked(tmp_path): + # F2: a plain command (first token has no "/") is not a path and is not + # existence-checked — "pytest -q" must stay clean. + from loop.contract import doctor_report + + target = _scaffold(tmp_path, "plain-cmd") + _set_task_verify(target, "pytest -q") + + report = doctor_report(target) + assert report["ok"] is True, report["issues"] + + +def test_f2_fresh_scaffold_stays_clean(tmp_path): + # F2 must not over-fire: a fresh scaffold has verify scripts and a resolving + # task verify, so it stays doctor-clean. + from loop.contract import doctor_report + + target = _scaffold(tmp_path, "fresh") + report = doctor_report(target) + assert report["ok"] is True, report["issues"] + + +def test_f7_file_target_resolves_to_owning_workspace(tmp_path): + # F7: a FILE target (loop doctor .loop/state.json or TASKS.json) resolved the + # workspace to the file itself, so every path underneath was garbage. A file + # target must resolve from its parent to the SAME paths as the dir target. + from loop.paths import resolve_loop_paths + + target = _scaffold(tmp_path, "file-target") + from_dir = resolve_loop_paths(target) + + from_state_file = resolve_loop_paths(target / ".loop" / "state.json") + assert from_state_file.workspace == from_dir.workspace + assert from_state_file.state == from_dir.state + assert from_state_file.loop_dir == from_dir.loop_dir + + from_tasks_file = resolve_loop_paths(target / "TASKS.json") + assert from_tasks_file.workspace == from_dir.workspace + assert from_tasks_file.tasks == from_dir.tasks + + +def test_f7_doctor_on_file_target_matches_dir_target(tmp_path): + # F7 at the CLI level: `doctor /.loop/state.json` must produce the same + # report as `doctor ` instead of a wall of garbage-path issues. + target = _scaffold(tmp_path, "file-cli") + + from_dir = _run_loop_cli("doctor", str(target)) + from_file = _run_loop_cli("doctor", str(target / ".loop" / "state.json")) + + assert from_dir.returncode == 0, from_dir.stderr + from_dir.stdout + assert from_file.returncode == 0, from_file.stderr + from_file.stdout + dir_report = json.loads(from_dir.stdout) + file_report = json.loads(from_file.stdout) + assert file_report["ok"] is True + assert file_report["paths"] == dir_report["paths"] + + def test_loop_doctor_flags_stub_verify_scripts(tmp_path): workspace = _write_valid_loop(tmp_path) (workspace / "scripts" / "verify-fast").write_text(