| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| import os |
| import subprocess |
| import sys |
| from pathlib import Path |
|
|
| import pytest |
|
|
| import nexum_runtime.executor as executor_module |
| from nexum_runtime.executor import execute_tool_call, list_tools |
| from nexum_runtime.tooling.artifacts import ArtifactStore |
| from nexum_runtime.tooling.contracts import ToolCall |
| from nexum_runtime.tooling.engineering import ( |
| DisclosureStore, |
| EvidenceBundleStore, |
| PatchStore, |
| ReproductionStore, |
| TriageStore, |
| ) |
| from nexum_runtime.tooling.security import ApprovalStore |
| from nexum_runtime.tooling.transactions import TransactionStore |
|
|
|
|
| def _successful_run( |
| _command: str, _cwd: str, _timeout_s: float |
| ) -> tuple[bool, str, str, int]: |
| return True, "stable output\n", "", 0 |
|
|
|
|
| def _triage_with_evidence( |
| workspace: Path, *, session_id: str = "engineering-session" |
| ) -> str: |
| reproduction, _stdout, _stderr = ReproductionStore(workspace).run( |
| command="inspect state", |
| working_directory=".", |
| snapshot_paths=(), |
| timeout_s=0.0, |
| session_id=session_id, |
| run_command=_successful_run, |
| ) |
| triage = TriageStore(workspace).create( |
| reproduction_ids=(reproduction.reproduction_id,), |
| observed_facts=("The contained run completed with stable output.",), |
| hypotheses=("The affected path may have stale state.",), |
| severity="moderate", |
| confidence=0.75, |
| next_actions=("Apply a guarded repair and verify it.",), |
| session_id=session_id, |
| ) |
| return triage.triage_id |
|
|
|
|
| def test_reproduction_receipts_compare_any_number_and_are_session_scoped( |
| tmp_path: Path, monkeypatch: pytest.MonkeyPatch |
| ) -> None: |
| secret = "reproduction-secret-value" |
| monkeypatch.setenv("SERVICE_TOKEN", secret) |
| target = tmp_path / "state.txt" |
| target.write_text("stable\n", encoding="utf-8") |
| store = ReproductionStore(tmp_path) |
|
|
| def run_with_secret( |
| _command: str, _cwd: str, _timeout_s: float |
| ) -> tuple[bool, str, str, int]: |
| return True, f"stable output {secret}\n", "", 0 |
|
|
| first, first_stdout, _ = store.run( |
| command="inspect state", |
| working_directory=".", |
| snapshot_paths=("state.txt",), |
| timeout_s=0.0, |
| session_id="session-one", |
| run_command=run_with_secret, |
| ) |
| second, second_stdout, _ = store.run( |
| command="inspect state", |
| working_directory=".", |
| snapshot_paths=("state.txt",), |
| timeout_s=0.0, |
| session_id="session-one", |
| run_command=run_with_secret, |
| ) |
| comparison = store.compare( |
| (first.reproduction_id, second.reproduction_id), |
| session_id="session-one", |
| ) |
|
|
| assert secret not in first_stdout |
| assert secret not in second_stdout |
| assert comparison["stable_outcome"] is True |
| assert comparison["same_before_state"] is True |
| assert comparison["same_after_state"] is True |
| with pytest.raises(PermissionError, match="does not belong"): |
| store.get(first.reproduction_id, session_id="session-two") |
|
|
|
|
| def test_triage_and_disclosure_preserve_evidence_without_private_output( |
| tmp_path: Path, monkeypatch: pytest.MonkeyPatch |
| ) -> None: |
| secret = "private-disclosure-value" |
| monkeypatch.setenv("PRIVATE_KEY", secret) |
| session_id = "disclosure-session" |
| reproduction, _stdout, _stderr = ReproductionStore(tmp_path).run( |
| command="inspect failure", |
| working_directory=".", |
| snapshot_paths=(), |
| timeout_s=0.0, |
| session_id=session_id, |
| run_command=lambda _command, _cwd, _timeout: ( |
| False, |
| "", |
| f"failure {secret} at {tmp_path}", |
| 2, |
| ), |
| ) |
| triage = TriageStore(tmp_path).create( |
| reproduction_ids=(reproduction.reproduction_id,), |
| observed_facts=(f"Failure observed at {tmp_path} with {secret}.",), |
| hypotheses=("Configuration may be inconsistent.",), |
| severity="high", |
| confidence=0.8, |
| next_actions=("Inspect the smallest affected surface.",), |
| session_id=session_id, |
| ) |
| disclosure = DisclosureStore(tmp_path).create( |
| triage_id=triage.triage_id, |
| title="Contained failure", |
| summary=f"A failure was reproduced at {tmp_path}.", |
| impact="The affected operation does not complete.", |
| remediation="Apply a guarded repair and verify the exact resulting state.", |
| audience="coordinated", |
| patch_id="", |
| session_id=session_id, |
| ) |
| _artifact, document = ArtifactStore(tmp_path).read( |
| disclosure.artifact_id, |
| session_id=session_id, |
| ) |
| text = document.decode("utf-8") |
|
|
| assert disclosure.evidence_sha256 in text |
| assert reproduction.reproduction_id in text |
| assert "## Observed Facts" in text |
| assert "## Hypotheses" in text |
| assert secret not in text |
| assert str(tmp_path) not in text |
| assert "[WORKSPACE]" in text |
| assert "Raw command output" in text |
|
|
|
|
| def test_patch_workflow_verifies_commits_and_can_restore_committed_bytes( |
| tmp_path: Path, |
| ) -> None: |
| session_id = "patch-session" |
| target = tmp_path / "value.txt" |
| target.write_text("before\n", encoding="utf-8") |
| triage_id = _triage_with_evidence(tmp_path, session_id=session_id) |
| store = PatchStore(tmp_path) |
| patch = store.begin( |
| triage_id=triage_id, |
| paths=("value.txt",), |
| session_id=session_id, |
| ) |
| before_sha256 = hashlib.sha256(b"before\n").hexdigest() |
| applied, error = store.apply( |
| patch.patch_id, |
| ( |
| { |
| "operation": "replace", |
| "path": "value.txt", |
| "expected_sha256": before_sha256, |
| "old_text": "before", |
| "new_text": "after", |
| }, |
| ), |
| session_id=session_id, |
| ) |
|
|
| assert error == "" |
| assert applied.status == "applied" |
| assert target.read_text(encoding="utf-8") == "after\n" |
|
|
| def verify( |
| _command: str, _cwd: str, _timeout_s: float |
| ) -> tuple[bool, str, str, int]: |
| assert target.read_text(encoding="utf-8") == "after\n" |
| return True, "verified\n", "", 0 |
|
|
| verified, receipt, _stdout, _stderr = store.verify( |
| patch.patch_id, |
| command="verify repair", |
| working_directory=".", |
| timeout_s=0.0, |
| session_id=session_id, |
| run_command=verify, |
| ) |
| assert verified.status == "verified" |
| assert receipt.reproduction_id in verified.verification_ids |
|
|
| committed = store.commit(patch.patch_id, session_id=session_id) |
| assert committed.status == "committed" |
| restored = store.rollback(patch.patch_id, session_id=session_id) |
| assert restored.status == "rolled_back" |
| assert target.read_text(encoding="utf-8") == "before\n" |
|
|
|
|
| def test_patch_application_is_all_or_nothing_on_stale_input(tmp_path: Path) -> None: |
| session_id = "atomic-patch-session" |
| first = tmp_path / "first.txt" |
| second = tmp_path / "second.txt" |
| first.write_text("first before\n", encoding="utf-8") |
| second.write_text("second before\n", encoding="utf-8") |
| triage_id = _triage_with_evidence(tmp_path, session_id=session_id) |
| store = PatchStore(tmp_path) |
| patch = store.begin( |
| triage_id=triage_id, |
| paths=("first.txt", "second.txt"), |
| session_id=session_id, |
| ) |
| failed, error = store.apply( |
| patch.patch_id, |
| ( |
| { |
| "operation": "replace", |
| "path": "first.txt", |
| "expected_sha256": hashlib.sha256(b"first before\n").hexdigest(), |
| "old_text": "first before", |
| "new_text": "first after", |
| }, |
| { |
| "operation": "replace", |
| "path": "second.txt", |
| "expected_sha256": "0" * 64, |
| "old_text": "second before", |
| "new_text": "second after", |
| }, |
| ), |
| session_id=session_id, |
| ) |
|
|
| assert failed.status == "rolled_back" |
| assert "ExpectedHashMismatch" in error |
| assert first.read_text(encoding="utf-8") == "first before\n" |
| assert second.read_text(encoding="utf-8") == "second before\n" |
|
|
|
|
| def test_executor_treats_observed_nonzero_reproduction_as_valid_evidence( |
| tmp_path: Path, monkeypatch: pytest.MonkeyPatch |
| ) -> None: |
| secret = "executor-secret-value" |
| monkeypatch.setenv("API_KEY", secret) |
| monkeypatch.setattr( |
| executor_module, |
| "_run_command", |
| lambda _command, _cwd, _timeout: ( |
| False, |
| f"reproduced {secret}\n", |
| "", |
| 7, |
| ), |
| ) |
| result = execute_tool_call( |
| ToolCall( |
| name="ReproductionRun", |
| args={ |
| "command": f"run --credential {secret}", |
| "snapshot_paths": [], |
| }, |
| raw="", |
| call_id="reproduction-call", |
| ), |
| cwd=str(tmp_path), |
| session_id="executor-session", |
| ) |
| payload = json.loads(result.output) |
|
|
| assert result.ok is True |
| assert result.exit_code == 7 |
| assert payload["record"]["command_ok"] is False |
| assert secret not in result.output |
| assert secret not in json.dumps(result.args) |
| assert "[REDACTED]" in json.dumps(result.args) |
|
|
| names = {row["name"] for row in list_tools()} |
| assert { |
| "ReproductionRun", |
| "ReproductionCompare", |
| "TriageCreate", |
| "TriageUpdate", |
| "TriageStatus", |
| "DisclosureCreate", |
| "DisclosureStatus", |
| "EvidenceBundleCreate", |
| "EvidenceBundleStatus", |
| "PatchBegin", |
| "PatchApply", |
| "PatchVerify", |
| "PatchCommit", |
| "PatchRollback", |
| } <= names |
|
|
|
|
| def test_triage_update_and_disclosure_status_tools(tmp_path: Path) -> None: |
| session_id = "triage-update-session" |
| first = _triage_with_evidence(tmp_path, session_id=session_id) |
| second_reproduction, _stdout, _stderr = ReproductionStore(tmp_path).run( |
| command="inspect second state", |
| working_directory=".", |
| snapshot_paths=(), |
| timeout_s=0.0, |
| session_id=session_id, |
| run_command=lambda _command, _cwd, _timeout: (False, "", "still failing", 3), |
| ) |
| update = execute_tool_call( |
| ToolCall( |
| name="TriageUpdate", |
| args={ |
| "triage_id": first, |
| "reproduction_ids": [second_reproduction.reproduction_id], |
| "observed_facts": ["Second contained run still fails."], |
| "hypotheses": ["Repair candidate did not touch the failing path."], |
| "severity": "high", |
| "confidence": 0.9, |
| "next_actions": ["Try the smaller patch path first."], |
| }, |
| raw="", |
| call_id="triage-update-call", |
| ), |
| cwd=str(tmp_path), |
| session_id=session_id, |
| ) |
| assert update.ok is True |
| updated = json.loads(update.output) |
| assert updated["severity"] == "high" |
| assert second_reproduction.reproduction_id in updated["reproduction_ids"] |
| assert "Second contained run still fails." in updated["observed_facts"] |
|
|
| disclosure = DisclosureStore(tmp_path).create( |
| triage_id=first, |
| title="Updated triage disclosure", |
| summary="Updated evidence is available.", |
| impact="The contained operation still fails.", |
| remediation="Use the next verified patch path.", |
| audience="maintainer", |
| patch_id="", |
| session_id=session_id, |
| ) |
| status = execute_tool_call( |
| ToolCall( |
| name="DisclosureStatus", |
| args={"disclosure_id": disclosure.disclosure_id}, |
| raw="", |
| call_id="disclosure-status-call", |
| ), |
| cwd=str(tmp_path), |
| session_id=session_id, |
| ) |
| assert status.ok is True |
| assert json.loads(status.output)["artifact_id"] == disclosure.artifact_id |
|
|
|
|
| def test_evidence_bundle_packages_sanitized_engineering_receipts( |
| tmp_path: Path, monkeypatch: pytest.MonkeyPatch |
| ) -> None: |
| session_id = "evidence-bundle-session" |
| secret = "bundle-secret-value" |
| monkeypatch.setenv("BUNDLE_TOKEN", secret) |
| target = tmp_path / "app.txt" |
| target.write_text("before\n", encoding="utf-8") |
|
|
| reproduction, _stdout, _stderr = ReproductionStore(tmp_path).run( |
| command=f"inspect --token {secret}", |
| working_directory=".", |
| snapshot_paths=("app.txt",), |
| timeout_s=0.0, |
| session_id=session_id, |
| run_command=lambda _command, _cwd, _timeout: ( |
| False, |
| f"observed {secret} at {tmp_path}", |
| "", |
| 2, |
| ), |
| ) |
| triage = TriageStore(tmp_path).create( |
| reproduction_ids=(reproduction.reproduction_id,), |
| observed_facts=(f"Failure reproduced at {tmp_path} with {secret}.",), |
| hypotheses=("The current value is stale.",), |
| severity="moderate", |
| confidence=0.7, |
| next_actions=("Apply a verified patch.",), |
| session_id=session_id, |
| ) |
| patch = PatchStore(tmp_path).begin( |
| triage_id=triage.triage_id, |
| paths=("app.txt",), |
| session_id=session_id, |
| ) |
| disclosure = DisclosureStore(tmp_path).create( |
| triage_id=triage.triage_id, |
| title="Maintainer handoff", |
| summary=f"Contained failure at {tmp_path}", |
| impact="The selected operation fails.", |
| remediation="Verify the smallest repair path.", |
| audience="maintainer", |
| patch_id="", |
| session_id=session_id, |
| ) |
| bundle = EvidenceBundleStore(tmp_path).create( |
| title=f"Handoff {tmp_path}", |
| reproduction_ids=(reproduction.reproduction_id,), |
| triage_ids=(triage.triage_id,), |
| disclosure_ids=(disclosure.disclosure_id,), |
| patch_ids=(patch.patch_id,), |
| session_id=session_id, |
| ) |
| _artifact, document = ArtifactStore(tmp_path).read( |
| bundle.artifact_id, |
| session_id=session_id, |
| ) |
| text = document.decode("utf-8") |
|
|
| assert bundle.bundle_id.startswith("evb_") |
| assert bundle.evidence_sha256 in text |
| assert reproduction.reproduction_id in text |
| assert triage.triage_id in text |
| assert disclosure.disclosure_id in text |
| assert patch.patch_id in text |
| assert secret not in text |
| assert str(tmp_path) not in text |
| assert "Raw command output" in text |
|
|
| status = execute_tool_call( |
| ToolCall( |
| name="EvidenceBundleStatus", |
| args={"bundle_id": bundle.bundle_id}, |
| raw="", |
| call_id="bundle-status-call", |
| ), |
| cwd=str(tmp_path), |
| session_id=session_id, |
| ) |
| assert status.ok is True |
| assert json.loads(status.output)["artifact_id"] == bundle.artifact_id |
|
|
| cross_session = execute_tool_call( |
| ToolCall( |
| name="EvidenceBundleStatus", |
| args={"bundle_id": bundle.bundle_id}, |
| raw="", |
| call_id="bundle-status-cross-session-call", |
| ), |
| cwd=str(tmp_path), |
| session_id="other-session", |
| ) |
| assert cross_session.ok is False |
| assert "does not belong to this session" in cross_session.error |
|
|
|
|
| def test_cli_can_resume_one_exact_approved_tool_action(tmp_path: Path) -> None: |
| target = tmp_path / "value.txt" |
| target.write_text("before", encoding="utf-8") |
| transaction = TransactionStore(tmp_path).begin( |
| ("value.txt",), session_id="cli-approval-session" |
| ) |
| target.write_text("after", encoding="utf-8") |
| current_sha256 = hashlib.sha256(b"after").hexdigest() |
| tool_text = ( |
| "TransactionRollback(" |
| f"transaction_id='{transaction.transaction_id}', " |
| f"expected_current={{'value.txt':'{current_sha256}'}})" |
| ) |
| source_root = Path(__file__).resolve().parents[1] / "src" |
| environment = os.environ.copy() |
| environment["PYTHONPATH"] = os.pathsep.join( |
| [str(source_root), environment.get("PYTHONPATH", "")] |
| ).rstrip(os.pathsep) |
| command = [ |
| sys.executable, |
| "-m", |
| "nexum_runtime.cli", |
| "tools", |
| "execute", |
| tool_text, |
| "--workspace", |
| str(tmp_path), |
| "--session-id", |
| "cli-approval-session", |
| "--call-id", |
| "rollback-call", |
| ] |
| requested = subprocess.run( |
| command, |
| text=True, |
| capture_output=True, |
| check=False, |
| env=environment, |
| ) |
| requested_payload = json.loads(requested.stdout) |
| approval_id = requested_payload["results"][0]["approval_id"] |
|
|
| assert requested.returncode == 1 |
| assert requested_payload["results"][0]["status"] == "input_required" |
| ApprovalStore(tmp_path).decide( |
| approval_id, |
| approved=True, |
| session_id="cli-approval-session", |
| ) |
| resumed = subprocess.run( |
| [*command, "--approval-id", approval_id], |
| text=True, |
| capture_output=True, |
| check=False, |
| env=environment, |
| ) |
| resumed_payload = json.loads(resumed.stdout) |
|
|
| assert resumed.returncode == 0 |
| assert resumed_payload["results"][0]["ok"] is True |
| assert target.read_text(encoding="utf-8") == "before" |
|
|
