diff --git a/docs/integrations/deepseek-harness-connector.md b/docs/integrations/deepseek-harness-connector.md index 1048239b7e..a34b591651 100644 --- a/docs/integrations/deepseek-harness-connector.md +++ b/docs/integrations/deepseek-harness-connector.md @@ -298,12 +298,13 @@ The repository includes four validation paths. The first three do not require th DeepSeek Harness SDK or a real dsh runtime: ```bash -python3 examples/dsh-turn-host-adapter-smoke.py +uv run --extra test python -m pytest tests/test_dsh_goal_mode.py python3 examples/loopx-turn-dsh-e2e-smoke.py python3 examples/loopx-turn-dsh-builtin-host-e2e-smoke.py ``` -The first guards adapter translation and result shaping. The second drives the +The first guards adapter translation, result shaping, and acceptance by the real +host-result validator. The second drives the full `loopx turn run-once -> adapter -> fake dsh -> validator -> writeback -> quota spend -> idempotent replay` chain. The third proves the built-in host's success path plus three bounded provider-capacity attempts, retry-budget diff --git a/examples/canary/smoke-fleet-health-smoke.py b/examples/canary/smoke-fleet-health-smoke.py deleted file mode 100644 index be21fe7239..0000000000 --- a/examples/canary/smoke-fleet-health-smoke.py +++ /dev/null @@ -1,61 +0,0 @@ -#!/usr/bin/env python3 -from __future__ import annotations - -import json -import sys -import tempfile -from pathlib import Path - - -REPO_ROOT = Path(__file__).resolve().parents[2] -sys.path.insert(0, str(REPO_ROOT)) - -from loopx.canary.smoke_health import build_smoke_fleet_health # noqa: E402 - - -def main() -> int: - inventory_payload = build_smoke_fleet_health(include_inventory=True) - inventory = inventory_payload["inventory"] - scripts = [entry["script"] for entry in inventory] - assert scripts - assert len(scripts) == len(set(scripts)) - assert inventory_payload["cadence_counts"]["daily_full_public"] == len(scripts) - assert inventory_payload["workflow_contract"]["missing_scripts"] == [] - - receipt = { - "schema_version": "canary_smoke_suite_run_v0", - "suite": "full-public", - "timeout_seconds": 120.0, - "failure_count": 0, - "timeout_count": 0, - "selected_checks": [ - { - "normalized": {"script": script}, - "status": "passed", - "ok": True, - "duration_seconds": 1.0, - } - for script in scripts - ], - } - with tempfile.TemporaryDirectory() as temp_dir: - receipt_path = Path(temp_dir) / "full-public.json" - receipt_path.write_text(json.dumps(receipt), encoding="utf-8") - health = build_smoke_fleet_health(receipt_paths=[receipt_path]) - - assert health["ok"] is True - assert health["ready"] is True - assert health["receipt_health"]["observed_script_count"] == len(scripts) - assert health["contract_reuse"]["semantic_duplicate_inference"] == "manual_review_required" - assert "inventory" not in health - assert len(json.dumps(health, ensure_ascii=False)) < 30_000 - print( - "smoke-fleet-health-smoke ok " - f"inventory={len(scripts)} owners={health['targeted_owner_count']} " - f"owner_gaps={health['owner_gap_count']}" - ) - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/examples/dsh-turn-host-adapter-smoke.py b/examples/dsh-turn-host-adapter-smoke.py deleted file mode 100755 index a82f9d74c4..0000000000 --- a/examples/dsh-turn-host-adapter-smoke.py +++ /dev/null @@ -1,358 +0,0 @@ -"""Smoke for the thin DeepSeek Harness Turn host adapter. - -Guards the reusable request/result translation without calling a model or the -network: signed authority extraction from the bounded Turn envelope, parsing of -the dsh final-message JSON candidate, result shaping, and acceptance by the -real LoopX host-result validator. -""" - -from __future__ import annotations - -from hashlib import sha256 -import importlib.util -import json -import subprocess -import sys -import tempfile -from pathlib import Path - -REPO_ROOT = Path(__file__).resolve().parents[1] -SCRIPTS = REPO_ROOT / "scripts" -for path in (str(REPO_ROOT), str(SCRIPTS)): - if path not in sys.path: - sys.path.insert(0, path) - -# The generic-cli e2e smoke owns the fake runner source every host smoke reuses, -# because a fake runner whose keyword signature lags the adapter turns into an -# unattributed ``host_failure`` instead of naming the seam that moved. -_BASE_PATH = REPO_ROOT / "examples" / "loopx-turn-dsh-e2e-smoke.py" -_SPEC = importlib.util.spec_from_file_location("loopx_turn_dsh_e2e_smoke", _BASE_PATH) -assert _SPEC is not None and _SPEC.loader is not None -_base = importlib.util.module_from_spec(_SPEC) -sys.modules[_SPEC.name] = _base -_SPEC.loader.exec_module(_base) - -import dsh_turn_host_adapter as adapter # noqa: E402 -from loopx.control_plane.turn_driver.executor import ( # noqa: E402 - validate_loopx_turn_host_result, -) -from loopx.control_plane.quota.turn_envelope import ( # noqa: E402 - turn_envelope_action_signature_document, -) - - -TURN_KEY = "sha256:" + "0" * 64 - - -def _request( - *, - recommended_action: str = "Do the small thing.", - primary_action: str = "Do the signed thing.", -) -> dict: - request = { - "schema_version": adapter.LOOPX_TURN_HOST_REQUEST_SCHEMA, - "turn_key": TURN_KEY, - "route": "primary_delivery", - "session": {"goal_id": "g", "agent_id": "a"}, - "turn_envelope": { - "schema_version": "loopx_turn_envelope_v0", - "goal_id": "g", - "agent_id": "a", - "action": { - "recommended_action": recommended_action, - "primary_action": primary_action, - "must_attempt": True, - }, - "required_reads": [ - { - "kind": "command", - "command": "git status --short", - "reason": "establish the workspace baseline", - } - ], - "boundary": { - "write_scope": ["docs/**", "tests/**"], - "workspace_guard": { - "schema_version": "workspace_guard_v0", - "status": "ready", - "action": "continue", - }, - }, - "action_signature": { - "schema_version": "loopx_action_signature_v0", - "matches": True, - }, - }, - "result_contract": { - "schema_version": adapter.LOOPX_TURN_RESULT_SCHEMA, - "completed_phases": list(adapter.COMPLETED_PHASES), - }, - } - signature = turn_envelope_action_signature_document(request["turn_envelope"]) - signature_hash = "sha256:" + sha256( - json.dumps( - signature, - ensure_ascii=False, - sort_keys=True, - separators=(",", ":"), - ).encode("utf-8") - ).hexdigest() - request["turn_envelope"]["action_signature"].update( - { - "source_hash": signature_hash, - "envelope_hash": signature_hash, - } - ) - return request - - -def _plan(request: dict) -> dict: - return { - "transaction": {"turn_key": request["turn_key"]}, - "turn_envelope": request["turn_envelope"], - } - - -def _assert(condition: bool, message: str) -> None: - if not condition: - raise AssertionError(message) - - -def test_action_text_uses_signed_primary_action() -> None: - request = _request( - recommended_action="legacy action must not win", - primary_action="signed primary action", - ) - _assert( - adapter.extract_action_text(request) == "signed primary action", - "signed primary_action must be the bounded task body", - ) - - -def test_unsigned_action_is_rejected() -> None: - request = _request() - request["turn_envelope"]["action_signature"]["matches"] = False - try: - adapter.extract_action_text(request) - except ValueError as exc: - _assert("action signature" in str(exc), "signature error must be actionable") - else: # pragma: no cover - defensive - raise AssertionError("unsigned TurnEnvelope action must fail closed") - - -def test_action_tampering_after_signature_is_rejected() -> None: - request = _request() - request["turn_envelope"]["action"]["primary_action"] = "tampered action" - try: - adapter.extract_action_text(request) - except ValueError as exc: - _assert("action signature" in str(exc), "tamper error must be actionable") - else: # pragma: no cover - defensive - raise AssertionError("post-signature TurnEnvelope tampering must fail closed") - - -def test_prompt_preserves_structured_turn_authority() -> None: - request = _request(recommended_action="legacy action must not appear") - authority = adapter.extract_turn_authority(request) - prompt = adapter.render_prompt(authority) - _assert( - authority["primary_action"] == "Do the signed thing.", - "primary_action must be preserved", - ) - _assert( - authority["required_reads"] == request["turn_envelope"]["required_reads"], - "required_reads must be preserved without prose reconstruction", - ) - _assert( - authority["write_scope"] == ["docs/**", "tests/**"], - "the complete write scope must be preserved", - ) - _assert( - authority["workspace_guard"] - == request["turn_envelope"]["boundary"]["workspace_guard"], - "the complete workspace guard must be preserved", - ) - _assert("legacy action must not appear" not in prompt, "legacy action must not leak") - for expected in ( - '"primary_action":"Do the signed thing."', - '"command":"git status --short"', - '"write_scope":["docs/**","tests/**"]', - '"workspace_guard":{"action":"continue"', - "- recommended_action:", - "- vision_unchanged_reason:", - ): - _assert(expected in prompt, f"prompt must contain structured authority: {expected}") - - -def test_parse_model_json_accepts_exact_and_fenced_json() -> None: - candidate = { - "result_kind": "validated_progress", - "classification": "updated config", - "recommended_action": "review", - "next_action": "run the smoke", - "vision_unchanged_reason": "the objective is unchanged", - "summary": "changed one field", - } - encoded = json.dumps(candidate) - _assert(adapter.parse_model_json(encoded) == candidate, "exact JSON must parse") - _assert( - adapter.parse_model_json("```json\n" + encoded + "\n```") == candidate, - "fenced JSON must parse", - ) - _assert( - adapter.parse_model_json("Here is the result:\n" + encoded + "\nDone") == candidate, - "JSON embedded in prose must parse", - ) - _assert(adapter.parse_model_json("not json") is None, "invalid output must fail closed") - - -def test_material_progress_result_validates() -> None: - request = _request() - candidate = { - "result_kind": "validated_progress", - "classification": "single surface change", - "summary": "edited one file", - "next_action": "review the diff", - } - result = adapter.build_result(request, candidate) - verdict = validate_loopx_turn_host_result(_plan(request), result) - _assert(verdict["ok"], "material result must validate: " + "; ".join(verdict["errors"])) - - -def test_wait_result_validates_without_material_fields() -> None: - request = _request() - result = adapter.build_result( - request, - { - "result_kind": "wait", - "classification": "throttled", - "next_action": "retry after cadence", - }, - ) - verdict = validate_loopx_turn_host_result(_plan(request), result) - _assert(verdict["ok"], "wait result must validate: " + "; ".join(verdict["errors"])) - _assert("delivery_batch_scale" not in result, "stop kinds carry no batch scale") - - -def test_missing_result_block_fails_closed_to_wait() -> None: - request = _request() - result = adapter.build_result(request, None, fallback_reason="no block found") - _assert(result["result_kind"] == "wait", "missing block must not fabricate progress") - verdict = validate_loopx_turn_host_result(_plan(request), result) - _assert(verdict["ok"], "fail-closed wait must validate: " + "; ".join(verdict["errors"])) - - -def test_unsupported_result_kind_fails_closed_to_wait() -> None: - request = _request() - result = adapter.build_result( - request, - {"result_kind": "not_a_loopx_kind", "summary": "bad"}, - ) - _assert(result["result_kind"] == "wait", "unsupported kind must not fabricate progress") - _assert( - result["classification"] == "unsupported_host_result_kind", - "unsupported kind must be labeled", - ) - verdict = validate_loopx_turn_host_result(_plan(request), result) - _assert(verdict["ok"], "unsupported-kind wait must validate: " + "; ".join(verdict["errors"])) - - -def test_text_fields_are_bounded() -> None: - request = _request() - candidate = { - "result_kind": "user_action_required", - "classification": "x" * 500, - "summary": "y" * 900, - "next_action": "z" * 5000, - } - result = adapter.build_result(request, candidate) - _assert( - len(result["classification"]) <= 120, - "classification must respect its limit", - ) - _assert(len(result["summary"]) <= 400, "summary must respect its limit") - _assert(len(result["next_action"]) <= 1_200, "next_action must respect its limit") - - -def test_subprocess_adapter_roundtrip_with_fake_dsh_runner() -> None: - """End-to-end: request -> fake dsh runner -> typed Turn result.""" - - with tempfile.TemporaryDirectory() as tmp: - root = Path(tmp) - workspace = root / "workspace" - workspace.mkdir() - runner = root / "fake_dsh_runner.py" - marker = root / "prompt.json" - block = json.dumps( - { - "result_kind": "validated_progress", - "classification": "did it", - "recommended_action": "review", - "summary": "one change", - "next_action": "review", - "vision_unchanged_reason": "the objective is unchanged", - } - ) - runner.write_text( - "import json, pathlib, sys\n" - + _base.FAKE_DSH_RUNNER_SIGNATURE - + f" pathlib.Path({str(marker)!r}).write_text(prompt, " - + "encoding='utf-8')\n" - + f" return {block!r}\n", - encoding="utf-8", - ) - - request = _request() - completed = subprocess.run( - [ - sys.executable, - str(SCRIPTS / "dsh_turn_host_adapter.py"), - "--dsh-runner", - str(runner), - "--workspace", - str(workspace), - ], - input=json.dumps(request), - text=True, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - check=False, - ) - _assert(completed.returncode == 0, "adapter must exit 0: " + completed.stderr) - result = json.loads(completed.stdout) - verdict = validate_loopx_turn_host_result(_plan(request), result) - _assert( - verdict["ok"], - "roundtrip result must validate: " + "; ".join(verdict["errors"]), - ) - _assert( - result["result_kind"] == "validated_progress", - "fake dsh result must carry the material result", - ) - prompt_text = marker.read_text(encoding="utf-8") - _assert("Do the signed thing." in prompt_text, "prompt must use signed action") - - -def main() -> int: - tests = [ - test_action_text_uses_signed_primary_action, - test_unsigned_action_is_rejected, - test_action_tampering_after_signature_is_rejected, - test_prompt_preserves_structured_turn_authority, - test_parse_model_json_accepts_exact_and_fenced_json, - test_material_progress_result_validates, - test_wait_result_validates_without_material_fields, - test_missing_result_block_fails_closed_to_wait, - test_unsupported_result_kind_fails_closed_to_wait, - test_text_fields_are_bounded, - test_subprocess_adapter_roundtrip_with_fake_dsh_runner, - ] - for test in tests: - test() - print(f"ok {test.__name__}") - print(f"\n{len(tests)} checks passed") - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/examples/traex-turn-host-adapter-smoke.py b/examples/traex-turn-host-adapter-smoke.py index 011fbf4c3d..5d278dde81 100644 --- a/examples/traex-turn-host-adapter-smoke.py +++ b/examples/traex-turn-host-adapter-smoke.py @@ -114,39 +114,6 @@ def _assert(condition: bool, message: str) -> None: raise AssertionError(message) -def test_action_text_uses_signed_primary_action() -> None: - request = _request( - recommended_action="legacy action must not win", - primary_action="signed primary action", - ) - _assert( - adapter.extract_action_text(request) == "signed primary action", - "signed primary_action must be the bounded task body", - ) - - -def test_unsigned_action_is_rejected() -> None: - request = _request() - request["turn_envelope"]["action_signature"]["matches"] = False - try: - adapter.extract_action_text(request) - except ValueError as exc: - _assert("action signature" in str(exc), "signature error must be actionable") - else: # pragma: no cover - defensive - raise AssertionError("unsigned TurnEnvelope action must fail closed") - - -def test_action_tampering_after_signature_is_rejected() -> None: - request = _request() - request["turn_envelope"]["action"]["primary_action"] = "tampered action" - try: - adapter.extract_action_text(request) - except ValueError as exc: - _assert("action signature" in str(exc), "tamper error must be actionable") - else: # pragma: no cover - defensive - raise AssertionError("post-signature TurnEnvelope tampering must fail closed") - - def test_prompt_preserves_structured_turn_authority() -> None: request = _request(recommended_action="legacy action must not appear") authority = adapter.extract_turn_authority(request) @@ -424,9 +391,6 @@ def test_timeout_terminates_traex_descendants() -> None: def main() -> int: tests = [ - test_action_text_uses_signed_primary_action, - test_unsigned_action_is_rejected, - test_action_tampering_after_signature_is_rejected, test_prompt_preserves_structured_turn_authority, test_structured_result_file_is_the_only_candidate_channel, test_material_progress_result_validates, diff --git a/loopx/dsh_goal_mode/README.md b/loopx/dsh_goal_mode/README.md index 4348896322..fb15ed3cd5 100644 --- a/loopx/dsh_goal_mode/README.md +++ b/loopx/dsh_goal_mode/README.md @@ -152,8 +152,8 @@ does not claim managed supervisor recovery, outer wake/timer ownership, or a cross-process resume guarantee. See `docs/integrations/deepseek-harness-connector.md` for the full connector -walkthrough and `examples/dsh-turn-host-adapter-smoke.py` plus -`examples/loopx-turn-dsh-e2e-smoke.py` for hermetic smokes. With the optional +walkthrough, `tests/test_dsh_goal_mode.py` for the adapter contract, and +`examples/loopx-turn-dsh-e2e-smoke.py` for the hermetic end-to-end smoke. With the optional SDK installed, run `examples/loopx-turn-dsh-real-e2e-smoke.py` once with `--host generic-cli` and once with `--host dsh`; both paths clear ambient DSH home variables and prove the explicit SDK-home wiring. diff --git a/tests/canary/test_smoke_fleet_health.py b/tests/canary/test_smoke_fleet_health.py index 52e175dcd1..6f62267dec 100644 --- a/tests/canary/test_smoke_fleet_health.py +++ b/tests/canary/test_smoke_fleet_health.py @@ -1,9 +1,14 @@ from __future__ import annotations +import contextlib +import io import json from pathlib import Path +import pytest + from loopx.canary.smoke_health import build_smoke_fleet_health +from loopx.cli import main as cli_main def _passing_receipt(scripts: list[str]) -> dict[str, object]: @@ -85,3 +90,45 @@ def test_failed_and_invalid_receipts_remain_distinct(tmp_path: Path) -> None: assert invalid["ok"] is False assert invalid["ready"] is False assert invalid["warnings"][0]["kind"] == "unsupported_receipt" + + +def _run_smoke_health_cli(receipt_dir: Path) -> tuple[int, dict[str, object]]: + # Mirrors the full-public workflow step: global --format json, then the + # shard directory as one --receipt argument. + output = io.StringIO() + with contextlib.redirect_stdout(output): + exit_code = cli_main( + ["--format", "json", "canary", "smoke-health", "--receipt", str(receipt_dir)] + ) + return exit_code, json.loads(output.getvalue()) + + +@pytest.mark.parametrize("failing_shard", [False, True], ids=["all_pass", "one_failure"]) +def test_cli_merges_a_shard_directory_into_the_workflow_ready_gate( + tmp_path: Path, failing_shard: bool +) -> None: + scripts = [ + entry["script"] + for entry in build_smoke_fleet_health(include_inventory=True)["inventory"] + ] + half = len(scripts) // 2 + receipt_dir = tmp_path / "smoke-results" + receipt_dir.mkdir() + first = _passing_receipt(scripts[:half]) + second = _passing_receipt(scripts[half:]) + if failing_shard: + second["failure_count"] = 1 + second["selected_checks"][0].update({"status": "failed", "ok": False}) + (receipt_dir / "shard-0.json").write_text(json.dumps(first), encoding="utf-8") + (receipt_dir / "shard-1.json").write_text(json.dumps(second), encoding="utf-8") + + exit_code, payload = _run_smoke_health_cli(receipt_dir) + + assert exit_code == 0 + receipt_health = payload["receipt_health"] + assert receipt_health["accepted_receipt_count"] == 2 + assert receipt_health["observed_script_count"] == len(scripts) + assert receipt_health["missing_script_count"] == 0 + assert payload["ready"] is (not failing_shard) + assert "inventory" not in payload + assert str(tmp_path) not in json.dumps(payload, ensure_ascii=False) diff --git a/tests/test_dsh_goal_mode.py b/tests/test_dsh_goal_mode.py index 1a2cec43d7..684b38dc30 100644 --- a/tests/test_dsh_goal_mode.py +++ b/tests/test_dsh_goal_mode.py @@ -15,6 +15,7 @@ from loopx.control_plane.quota.turn_envelope import ( turn_envelope_action_signature_document, ) +from loopx.control_plane.turn_driver.executor import validate_loopx_turn_host_result from loopx.control_plane.turn_driver.host_failure import ( BuiltInHostError, build_host_failure_record, @@ -363,6 +364,26 @@ def test_prompt_requests_one_typed_public_safe_json_result() -> None: assert "credentials" in prompt +def test_prompt_carries_the_signed_authority_without_prose_reconstruction() -> None: + request = _signed_request() + authority = turn_host_adapter.extract_turn_authority(request) + envelope = request["turn_envelope"] + assert authority["primary_action"] == "Do the signed thing." + assert authority["required_reads"] == envelope["required_reads"] + assert authority["write_scope"] == envelope["boundary"]["write_scope"] + assert authority["workspace_guard"] == envelope["boundary"]["workspace_guard"] + + prompt = turn_host_adapter.render_prompt(authority) + assert "legacy action must not win" not in prompt + for expected in ( + '"primary_action":"Do the signed thing."', + '"command":"git status --short"', + '"write_scope":["docs/**","tests/**"]', + '"workspace_guard":{"action":"continue"', + ): + assert expected in prompt, expected + + def test_parse_model_json_tolerates_prose_and_fences() -> None: payload = {"result_kind": "wait", "summary": "nothing to do"} assert turn_host_adapter.parse_model_json(json.dumps(payload)) == payload @@ -429,6 +450,68 @@ def test_build_result_shapes_material_results_with_required_fields() -> None: assert result["vision_unchanged_reason"] +def _turn_plan(request: dict) -> dict: + return { + "transaction": {"turn_key": request["turn_key"]}, + "turn_envelope": request["turn_envelope"], + } + + +@pytest.mark.parametrize( + ("candidate", "expected_kind"), + [ + ( + { + "result_kind": "validated_progress", + "classification": "single surface change", + "summary": "edited one file", + "next_action": "review the diff", + }, + "validated_progress", + ), + ( + { + "result_kind": "wait", + "classification": "throttled", + "next_action": "retry after cadence", + }, + "wait", + ), + (None, "wait"), + ({"result_kind": "not_a_loopx_kind", "summary": "bad"}, "wait"), + ], + ids=["material", "wait", "missing_block", "unsupported_kind"], +) +def test_build_result_is_accepted_by_the_real_host_result_validator( + candidate: dict | None, expected_kind: str +) -> None: + # Shaping alone is not enough: every adapter output, including fail-closed + # waits, must pass the same validator the Turn executor applies. + request = _signed_request() + result = turn_host_adapter.build_result(request, candidate) + assert result["result_kind"] == expected_kind + if expected_kind == "wait": + assert "delivery_batch_scale" not in result + verdict = validate_loopx_turn_host_result(_turn_plan(request), result) + assert verdict["ok"], verdict["errors"] + + +def test_build_result_bounds_every_free_text_field() -> None: + result = turn_host_adapter.build_result( + _signed_request(), + { + "result_kind": "user_action_required", + "classification": "x" * 500, + "summary": "y" * 900, + "next_action": "z" * 5000, + }, + ) + # Independent limits from the Turn result contract, not the adapter's table. + assert len(result["classification"]) <= 120 + assert len(result["summary"]) <= 400 + assert len(result["next_action"]) <= 1_200 + + def test_adapter_runs_hermetically_through_the_module_entry() -> None: # `python -m loopx.dsh_goal_mode` must drive the same stdin/stdout # contract as the legacy script, using a fake dsh runner.