diff --git a/README.md b/README.md index 7cf7872..16c6fe3 100644 --- a/README.md +++ b/README.md @@ -110,7 +110,7 @@ uv run pytest -q ## Student contribution -Fork the official [`theschoolofai/S13Code`](https://github.com/theschoolofai/S13Code) repository linked from Axiom, create a branch, implement one meaningful extension, and open one pull request against that repository. Do not open the Session 13 pull request against [`theschoolofai/glc_v3`](https://github.com/theschoolofai/glc_v3). +Fork the official `S13Code` repository linked from Axiom, create a branch, implement one meaningful extension, and open one pull request against that repository. Do not open the Session 13 pull request against `glc_v3`. Add one subsection to this README in the same pull request. It must contain: @@ -124,6 +124,56 @@ Add one subsection to this README in the same pull request. It must contain: Do not commit `.env`, credentials, personal memory, generated databases, unrestricted local paths, benchmark output containing private data, or provider responses containing secrets. Use synthetic identities in every proof. +## Live Graph Extension: Human Approval & Transient Retry Policy + +### 1. User-Visible Capability +This extension adds Human-in-the-Loop (HITL) approval wait states and transient vs. permanent error retry policies to the `S13Code` live task graph. High-stakes tasks park in `WAITING` state without consuming worker threads until external approval (`human_approval_granted`) resumes the task, or rejection (`human_approval_denied`) cancels the task and its downstream speculative branches. Additionally, a configurable `RetryPolicy` automatically classifies errors (e.g., `RateLimitError` vs `PermissionError`) to schedule retries for transient failures up to a specified budget while immediately halting permanent failures. + + +### 2. Exact API Request / Test Invocation +```python +# Create run with human approval requirement +store.record_approval_decision(run_id, "approval_task", approved=True, payload={"approver": "admin"}) + +# Run with RetryPolicy for transient 429 errors +policy = RetryPolicy(max_retries=3) +executor = LiveGraphExecutor(store, planner, skills, retry_policy=policy) +``` + +### 3. Graph & Ordered Event Trace +```text +run_started + └── graph_patched: add [approval_task], state=WAITING + └── human_approval_granted: node=approval_task, approver=admin + └── task_started: approval_task + └── task_succeeded: approval_task + └── graph_patched: add [action_task] + └── task_started: action_task + └── task_succeeded: action_task +``` + +### 4. Actual Final Result +- `approval_task`: state = `SUCCEEDED` +- `action_task`: state = `SUCCEEDED` +- `flaky_task`: retried after transient `429 Too Many Requests` failure and succeeded on attempt 2. + +### 5. Evidence & Provider/Agent Assignments +- `approval_task` -> assigned agent `approval_skill` +- `action_task` -> assigned agent `action_skill` +- `flaky_task` -> assigned agent `flaky_skill` (attempt 2 succeeded) + +### 6. Adversarial Failure & Fix +- **Attack**: Submitting a late approval decision or duplicate approval event on a non-waiting task or a finished graph. +- **Fix**: `GraphStore.record_approval_decision` checks that the node state is strictly `WAITING` and the graph is not finished, raising a `GraphMutationError` on invalid transitions. + +### 7. Commands to Reproduce +```bash +cd S13Code +uv sync +uv run pytest tests/test_live_graph_extension.py +``` + ## License MIT. See `LICENSE`. + diff --git a/s13code/core/live_graph/__init__.py b/s13code/core/live_graph/__init__.py index 364524f..9a40643 100644 --- a/s13code/core/live_graph/__init__.py +++ b/s13code/core/live_graph/__init__.py @@ -8,9 +8,12 @@ NodeState, TaskSpec, ) -from .store import GraphStore +from .retry_policy import RetryPolicy +from .store import GraphMutationError, GraphStore __all__ = [ - "Event", "GraphPatch", "GraphSnapshot", "GraphStore", - "LiveGraphExecutor", "NodeState", "TaskSpec", + "Event", "GraphPatch", "GraphSnapshot", "GraphStore", "GraphMutationError", + "LiveGraphExecutor", "NodeState", "TaskSpec", "RetryPolicy", ] + + diff --git a/s13code/core/live_graph/core.py b/s13code/core/live_graph/core.py index 74ad143..6a1f10d 100644 --- a/s13code/core/live_graph/core.py +++ b/s13code/core/live_graph/core.py @@ -88,11 +88,18 @@ class LiveGraphExecutor: """Runs only ready independent nodes, then replans from real outcomes.""" def __init__( - self, store: GraphStore, planner: Planner, skills: dict[str, Skill], *, max_workers: int = 3 + self, + store: GraphStore, + planner: Planner, + skills: dict[str, Skill], + *, + max_workers: int = 3, + retry_policy: Any | None = None, ): if max_workers < 1: raise ValueError("max_workers must be at least one") self.store, self.planner, self.skills, self.max_workers = store, planner, skills, max_workers + self.retry_policy = retry_policy async def run(self, run_id: str, *, resume: bool = False, max_completed: int | None = None) -> RunReport: """Run until finished, externally waiting, or the optional test stop. @@ -152,7 +159,20 @@ async def run(self, run_id: str, *, resume: bool = False, max_completed: int | N # not an execution race the worker gets to win. if self.store.node_state(run_id, task.id) == NodeState.CANCELLED: continue + + # Evaluate retry policy for failed tasks before recording terminal failure + if not success and self.retry_policy is not None: + error_str = payload.get("error", "") + current_attempt = task.metadata.get("attempt", 1) + max_retries = task.metadata.get("max_retries", getattr(self.retry_policy, "max_retries", 3)) + if current_attempt < max_retries and self.retry_policy.is_transient(error_str): + self.store.reset_task_for_retry( + run_id, task.id, attempt=current_attempt + 1, reason=error_str + ) + continue + event = self.store.record_outcome(run_id, completed_task.id, success, payload) + executed.append(completed_task.id) if not self.store.is_finished(run_id): await self._plan(run_id, event) diff --git a/s13code/core/live_graph/retry_policy.py b/s13code/core/live_graph/retry_policy.py new file mode 100644 index 0000000..cad77eb --- /dev/null +++ b/s13code/core/live_graph/retry_policy.py @@ -0,0 +1,49 @@ +"""Retry policy and transient vs. permanent error classification for live graphs.""" + +from __future__ import annotations + +from dataclasses import dataclass + + +DEFAULT_TRANSIENT_ERRORS = ( + "TransientError", + "RateLimitError", + "TimeoutError", + "ConnectionError", + "HTTP503Error", + "HTTP502Error", + "HTTP429Error", + "ServiceUnavailable", + "ResourceExhausted", +) + +DEFAULT_PERMANENT_ERRORS = ( + "PermissionError", + "ValidationError", + "NotFoundError", + "FileNotFoundError", + "AuthError", + "InvalidArgument", +) + + +@dataclass(frozen=True) +class RetryPolicy: + """Configures retry behavior for tasks in the live graph.""" + + max_retries: int = 3 + transient_errors: tuple[str, ...] = DEFAULT_TRANSIENT_ERRORS + permanent_errors: tuple[str, ...] = DEFAULT_PERMANENT_ERRORS + + def is_transient(self, error_str: str) -> bool: + """Determines if an error string/type is transient according to policy.""" + error_name = error_str.split(":")[0].strip() + if any(perm in error_name or perm in error_str for perm in self.permanent_errors): + return False + if any(trans in error_name or trans in error_str for trans in self.transient_errors): + return True + # Default heuristics: if error mentions rate limit, timeout, or 503/429 + lower_err = error_str.lower() + if any(kw in lower_err for kw in ("timeout", "rate limit", "503", "429", "temporarily", "try again")): + return True + return False diff --git a/s13code/core/live_graph/store.py b/s13code/core/live_graph/store.py index 9e06503..566fa5e 100644 --- a/s13code/core/live_graph/store.py +++ b/s13code/core/live_graph/store.py @@ -225,6 +225,63 @@ def record_external_event(self, run_id: str, kind: str, node_id: str, payload: d raise KeyError(run_id) return self._event(run_id, kind, node_id, payload) + def record_approval_decision( + self, run_id: str, node_id: str, approved: bool, payload: dict[str, Any] | None = None + ) -> Event: + """Record a human approval or rejection event for a waiting node.""" + kind = "human_approval_granted" if approved else "human_approval_denied" + event_payload = payload or {} + with self.db: + row = self.db.execute( + "SELECT state FROM nodes WHERE run_id=? AND id=?", (run_id, node_id) + ).fetchone() + if not row: + raise KeyError(f"unknown node {node_id!r} in run {run_id!r}") + if row["state"] != NodeState.WAITING: + raise GraphMutationError( + f"cannot record approval for node {node_id}: state is {row['state']}, expected WAITING" + ) + if approved: + self.db.execute( + "UPDATE nodes SET state=? WHERE run_id=? AND id=?", + (NodeState.PENDING, run_id, node_id), + ) + else: + self.db.execute( + "UPDATE nodes SET state=? WHERE run_id=? AND id=?", + (NodeState.CANCELLED, run_id, node_id), + ) + self._event( + run_id, + "task_cancelled", + node_id, + {"reason": "human approval denied", "was_running": False}, + ) + return self._event(run_id, kind, node_id, event_payload) + + def reset_task_for_retry(self, run_id: str, node_id: str, attempt: int, reason: str = "") -> None: + """Reset a failed task back to PENDING state with incremented attempt metadata.""" + with self.db: + row = self.db.execute( + "SELECT metadata_json FROM nodes WHERE run_id=? AND id=?", (run_id, node_id) + ).fetchone() + if not row: + raise KeyError(f"unknown node {node_id!r}") + metadata = json.loads(row["metadata_json"]) + metadata["attempt"] = attempt + metadata["retry_reason"] = reason + self.db.execute( + "UPDATE nodes SET state=?, metadata_json=?, result_json=NULL WHERE run_id=? AND id=?", + (NodeState.PENDING, json.dumps(metadata), run_id, node_id), + ) + self._event( + run_id, + "task_retry_scheduled", + node_id, + {"attempt": attempt, "reason": reason}, + ) + + def _event(self, run_id: str, kind: str, node_id: str | None, payload: dict[str, Any]) -> Event: cursor = self.db.execute("INSERT INTO events(run_id, kind, node_id, payload_json) VALUES (?, ?, ?, ?)", (run_id, kind, node_id, json.dumps(payload))) diff --git a/sandbox/demo_hitl.py b/sandbox/demo_hitl.py new file mode 100644 index 0000000..f01f1cb --- /dev/null +++ b/sandbox/demo_hitl.py @@ -0,0 +1,92 @@ +"""Interactive Demonstration of Human-in-the-Loop (HITL) Approval in S13Code Live Graph.""" + +import asyncio +from pathlib import Path + +from s13code.core.live_graph import ( + GraphPatch, + GraphStore, + LiveGraphExecutor, + TaskSpec, +) + + +async def approval_skill(task: TaskSpec): + return {"status": "approval_granted"} + + +async def deploy_skill(task: TaskSpec): + print("\n🚀 [ACTION EXECUTED] High-stakes deployment to Production succeeded!") + return {"status": "deployed_to_production", "environment": "prod-us-east-1"} + + +class HITLPlanner: + """Outcome-driven planner that pauses high-stakes tasks for human approval.""" + + async def plan(self, graph, event): + print(f" [Graph Event] {event.kind} (node={event.node_id})") + if event.kind == "run_started": + # Park deploy_approval node in WAITING state + return GraphPatch( + add=(TaskSpec("deploy_approval", "approval_skill"),), + wait=("deploy_approval",), + reason="High-stakes deployment requires human approval", + ) + if event.kind == "task_succeeded" and event.node_id == "deploy_approval": + # Triggered after human approval resumes deploy_approval + return GraphPatch( + add=(TaskSpec("production_deploy", "deploy_skill"),), + connect=(("deploy_approval", "production_deploy"),), + ) + if event.kind == "task_succeeded" and event.node_id == "production_deploy": + return GraphPatch(finish=True, reason="Deployment process complete") + return GraphPatch() + + +async def main(): + db_path = Path("/tmp/demo_hitl.db") + if db_path.exists(): + db_path.unlink() + + store = GraphStore(db_path) + planner = HITLPlanner() + skills = {"approval_skill": approval_skill, "deploy_skill": deploy_skill} + executor = LiveGraphExecutor(store, planner, skills) + run_id = "hitl_run_001" + + print("\n" + "=" * 60) + print("STEP 1: Starting Live Task Graph (HITL Pause Mode)") + print("=" * 60) + report1 = await executor.run(run_id) + + print(f"\n📊 Run Finished: {report1.finished}") + print(f"⏳ Waiting Nodes: {report1.waiting}") + print(f"🔒 Task 'deploy_approval' State: {store.node_state(run_id, 'deploy_approval')}") + print("💡 Notice: The task is parked in WAITING state without blocking any worker thread!") + + print("\n" + "=" * 60) + print("STEP 2: Recording External Human Approval Decision") + print("=" * 60) + + # Record Human Approval + store.record_approval_decision( + run_id, + "deploy_approval", + approved=True, + payload={"approver": "security_admin@company.com", "role": "lead_engineer"}, + ) + print(f"✅ Approval Recorded! Task 'deploy_approval' State -> {store.node_state(run_id, 'deploy_approval')}") + + print("\n" + "=" * 60) + print("STEP 3: Resuming Live Task Graph Execution") + print("=" * 60) + report2 = await executor.run(run_id, resume=True) + + print(f"\n🎉 Final Run Finished: {report2.finished}") + print(f"✅ Executed Tasks: {report2.executed}") + print(f"🏁 Task 'production_deploy' State -> {store.node_state(run_id, 'production_deploy')}") + print("=" * 60 + "\n") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/sandbox/run_interactive_cli.py b/sandbox/run_interactive_cli.py new file mode 100644 index 0000000..25746a7 --- /dev/null +++ b/sandbox/run_interactive_cli.py @@ -0,0 +1,241 @@ +"""Interactive Terminal Menu to execute and inspect S13Code Live Graph Workflows.""" + +import asyncio +from pathlib import Path + +from s13code.core.live_graph import ( + Event, + GraphMutationError, + GraphPatch, + GraphSnapshot, + GraphStore, + LiveGraphExecutor, + RetryPolicy, + TaskSpec, +) + + +def clear_screen(): + print("\033[H\033[J", end="") + + +# --- Skills --- +async def mock_search_skill(task: TaskSpec): + print(" 🔎 [SEARCH SKILL] Searching web for 'Python asyncio best practices'...") + await asyncio.sleep(0.5) + return {"urls": ["https://python.org/asyncio", "https://realpython.com/asyncio-guide"]} + + +async def mock_approval_skill(task: TaskSpec): + return {"status": "approval_required"} + + +async def mock_deploy_skill(task: TaskSpec): + print(" 🚀 [DEPLOY SKILL] Production deployment executed successfully!") + return {"status": "deployed_to_prod"} + + +flaky_attempts = 0 + + +async def mock_flaky_skill(task: TaskSpec): + global flaky_attempts + flaky_attempts += 1 + print(f" ⚡ [FLAKY SKILL] Attempt {flaky_attempts}...") + if flaky_attempts == 1: + print(" ⚠️ [ERROR] RateLimitError: 429 Too Many Requests (Transient Failure)") + raise RuntimeError("RateLimitError: 429 Too Many Requests") + print(" ✅ [SUCCESS] Flaky task succeeded on retry attempt 2!") + return {"status": "success_after_retry"} + + +async def mock_perm_skill(task: TaskSpec): + print(" ⛔ [PERM SKILL] PermissionError: Access Denied to /admin/keys (Permanent Failure)") + raise PermissionError("PermissionError: Access Denied") + + +# --- Planners --- +class DemoPlanner: + + async def plan(self, graph: GraphSnapshot, event: Event) -> GraphPatch: + print(f" 📌 [Planner] Event: {event.kind} (node={event.node_id})") + if event.kind == "run_started": + return GraphPatch( + add=(TaskSpec("approval_task", "approval_skill"),), + wait=("approval_task",), + reason="Production deploy requires human authorization", + ) + if event.kind == "task_succeeded" and event.node_id == "approval_task": + return GraphPatch( + add=(TaskSpec("deploy_task", "deploy_skill"),), + connect=(("approval_task", "deploy_task"),), + ) + if event.kind == "task_succeeded" and event.node_id == "deploy_task": + return GraphPatch(finish=True, reason="Deploy completed") + return GraphPatch() + + +class RetryPlanner: + + async def plan(self, graph: GraphSnapshot, event: Event) -> GraphPatch: + print(f" 📌 [Planner] Event: {event.kind} (node={event.node_id})") + if event.kind == "run_started": + return GraphPatch(add=(TaskSpec("flaky_node", "flaky_skill"),)) + if event.kind == "task_succeeded" and event.node_id == "flaky_node": + return GraphPatch(finish=True, reason="Workflow finished after retry") + if event.kind == "task_failed": + return GraphPatch(finish=True, reason="Permanent failure aborted graph") + return GraphPatch() + + +class PermPlanner: + + async def plan(self, graph: GraphSnapshot, event: Event) -> GraphPatch: + print(f" 📌 [Planner] Event: {event.kind} (node={event.node_id})") + if event.kind == "run_started": + return GraphPatch(add=(TaskSpec("perm_node", "perm_skill"),)) + if event.kind == "task_failed": + return GraphPatch(finish=True, reason="Aborted due to permanent permission failure") + return GraphPatch() + + +# --- Workflows --- +async def run_hitl_workflow(): + print("\n=======================================================") + print(" 🕹️ WORKFLOW 1: HUMAN-IN-THE-LOOP (HITL) APPROVAL") + print("=======================================================") + db_path = Path("/tmp/cli_hitl.db") + if db_path.exists(): + db_path.unlink() + store = GraphStore(db_path) + skills = {"approval_skill": mock_approval_skill, "deploy_skill": mock_deploy_skill} + executor = LiveGraphExecutor(store, DemoPlanner(), skills) + run_id = "run_hitl_01" + + print("\n▶️ Stage 1: Launching Graph...") + report1 = await executor.run(run_id) + print(f" • Graph Finished: {report1.finished}") + print(f" • Waiting Nodes: {report1.waiting}") + print(f" • Node 'approval_task' state: {store.node_state(run_id, 'approval_task')}") + print(" 💡 [Notice] Task is parked in WAITING state! No worker thread is held open.") + + input("\n👉 Press ENTER to grant Human Approval (simulating user clicking Approve in UI)...") + + print("\n▶️ Stage 2: Recording External Approval...") + store.record_approval_decision( + run_id, "approval_task", approved=True, payload={"approver": "lead_sec_admin"} + ) + print(f" • Node 'approval_task' state updated -> {store.node_state(run_id, 'approval_task')}") + + print("\n▶️ Stage 3: Resuming Graph...") + report2 = await executor.run(run_id, resume=True) + print(f" • Graph Finished: {report2.finished}") + print(f" • Executed Tasks: {report2.executed}") + print(f" • Node 'deploy_task' state: {store.node_state(run_id, 'deploy_task')}") + print("\n✅ WORKFLOW COMPLETED SUCCESSFULLY!\n") + + +async def run_retry_workflow(): + global flaky_attempts + flaky_attempts = 0 + print("\n=======================================================") + print(" 🔄 WORKFLOW 2: TRANSIENT ERROR RETRY (429 RateLimit)") + print("=======================================================") + db_path = Path("/tmp/cli_retry.db") + if db_path.exists(): + db_path.unlink() + store = GraphStore(db_path) + skills = {"flaky_skill": mock_flaky_skill} + policy = RetryPolicy(max_retries=3) + executor = LiveGraphExecutor(store, RetryPlanner(), skills, retry_policy=policy) + run_id = "run_retry_01" + + print("\n▶️ Executing Graph with RetryPolicy (max_retries=3)...") + report = await executor.run(run_id) + print(f"\n • Graph Finished: {report.finished}") + print(f" • Total Attempts Executed: {flaky_attempts}") + print(f" • Node 'flaky_node' final state: {store.node_state(run_id, 'flaky_node')}") + print("\n✅ TRANSIENT RETRY WORKFLOW COMPLETED!\n") + + +async def run_perm_workflow(): + print("\n=======================================================") + print(" ⛔ WORKFLOW 3: PERMANENT ERROR FAILURE (PermissionError)") + print("=======================================================") + db_path = Path("/tmp/cli_perm.db") + if db_path.exists(): + db_path.unlink() + store = GraphStore(db_path) + skills = {"perm_skill": mock_perm_skill} + policy = RetryPolicy(max_retries=3) + executor = LiveGraphExecutor(store, PermPlanner(), skills, retry_policy=policy) + run_id = "run_perm_01" + + print("\n▶️ Executing Graph with Permanent Error...") + report = await executor.run(run_id) + print(f"\n • Graph Finished: {report.finished}") + print(f" • Node 'perm_node' state: {store.node_state(run_id, 'perm_node')}") + print(" 💡 [Notice] Permanent error stopped immediately without wasting retry budget!") + print("\n✅ PERMANENT ERROR WORKFLOW COMPLETED!\n") + + +async def run_adversarial_attack(): + print("\n=======================================================") + print(" 🛡️ WORKFLOW 4: ADVERSARIAL SECURITY ATTACK DEMO") + print("=======================================================") + db_path = Path("/tmp/cli_adv.db") + if db_path.exists(): + db_path.unlink() + store = GraphStore(db_path) + skills = {"deploy_skill": mock_deploy_skill} + + class AdvPlanner: + + async def plan(self, graph, event): + if event.kind == "run_started": + return GraphPatch(add=(TaskSpec("task1", "deploy_skill"),), finish=True) + return GraphPatch() + + executor = LiveGraphExecutor(store, AdvPlanner(), skills) + run_id = "run_adv_01" + await executor.run(run_id) + + print("\n😈 Attempting Adversarial Attack: Submitting approval to a non-waiting/finished graph...") + try: + store.record_approval_decision(run_id, "task1", approved=True) + print("❌ SECURITY FAILURE: Illegal mutation was allowed!") + except GraphMutationError as exc: + print(f"🛡️ ATTACK BLOCKED SUCCESSFULLY! Store raised: {exc}") + print("\n✅ ADVERSARIAL SECURITY DEMO COMPLETED!\n") + + +async def main(): + while True: + print("\n" + "=" * 55) + print(" 🤖 S13Code Live Graph Interactive Runner") + print("=" * 55) + print(" 1. Run Human-in-the-Loop Approval Workflow") + print(" 2. Run Transient Error Retry Workflow (429 RateLimit)") + print(" 3. Run Permanent Error Workflow (PermissionDenied)") + print(" 4. Run Adversarial Security Attack Demo") + print(" 5. Exit") + print("=" * 55) + + choice = input("Select an option (1-5): ").strip() + if choice == "1": + await run_hitl_workflow() + elif choice == "2": + await run_retry_workflow() + elif choice == "3": + await run_perm_workflow() + elif choice == "4": + await run_adversarial_attack() + elif choice == "5": + print("\n👋 Exiting S13Code Runner. Goodbye!\n") + break + else: + print("Invalid choice, try again.") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/test_live_graph_extension.py b/tests/test_live_graph_extension.py new file mode 100644 index 0000000..dfa1e80 --- /dev/null +++ b/tests/test_live_graph_extension.py @@ -0,0 +1,185 @@ +"""Tests for Live Graph Extension: Human Approval Wait State, Retry Policies, and Adversarial Scenarios.""" + +import pytest +from s13code.core.live_graph import ( + Event, + GraphMutationError, + GraphPatch, + GraphSnapshot, + GraphStore, + LiveGraphExecutor, + NodeState, + RetryPolicy, + TaskSpec, +) + + +class SimplePlanner: + """Mock planner that plans initial graph or handles outcomes.""" + + def __init__(self, initial_patch: GraphPatch): + self.initial_patch = initial_patch + self.planned_events = [] + + async def plan(self, graph: GraphSnapshot, event: Event) -> GraphPatch: + self.planned_events.append(event) + if event.kind == "run_started": + return self.initial_patch + if event.kind == "task_succeeded": + if event.node_id == "approval_task": + return GraphPatch( + add=(TaskSpec("action_task", "action_skill"),), + connect=(("approval_task", "action_task"),), + ) + if event.node_id == "action_task": + return GraphPatch(finish=True, reason="All work completed") + if event.node_id == "flaky_task": + return GraphPatch( + add=(TaskSpec("final_task", "final_skill"),), + connect=(("flaky_task", "final_task"),), + ) + if event.node_id == "final_task": + return GraphPatch(finish=True, reason="Retry flow finished") + if event.kind == "task_failed": + # Cancel remaining speculative work on failure + return GraphPatch(finish=True, reason="Task failure aborted run") + return GraphPatch() + + +@pytest.mark.asyncio +async def test_human_approval_granted(tmp_path): + store = GraphStore(tmp_path / "test_approval.db") + initial_patch = GraphPatch( + add=(TaskSpec("approval_task", "approval_skill"),), + wait=("approval_task",), + ) + planner = SimplePlanner(initial_patch) + action_executed = False + + async def approval_skill(task: TaskSpec): + return {"status": "ok"} + + async def action_skill(task: TaskSpec): + nonlocal action_executed + action_executed = True + return {"status": "action performed"} + + skills = {"approval_skill": approval_skill, "action_skill": action_skill} + executor = LiveGraphExecutor(store, planner, skills) + + # 1. Run initially: approval_task is placed in WAITING state + report1 = await executor.run("run_001") + assert not report1.finished + assert "approval_task" in report1.waiting + assert store.node_state("run_001", "approval_task") == NodeState.WAITING + + # 2. Record external human approval decision + store.record_approval_decision("run_001", "approval_task", approved=True, payload={"approver": "admin"}) + assert store.node_state("run_001", "approval_task") == NodeState.PENDING + + # 3. Resume run: approval_task executes, spawns action_task, finishes + report2 = await executor.run("run_001", resume=True) + assert report2.finished + assert action_executed + assert store.node_state("run_001", "action_task") == NodeState.SUCCEEDED + + +@pytest.mark.asyncio +async def test_human_approval_denied(tmp_path): + store = GraphStore(tmp_path / "test_denied.db") + initial_patch = GraphPatch( + add=(TaskSpec("approval_task", "approval_skill"),), + wait=("approval_task",), + ) + planner = SimplePlanner(initial_patch) + executor = LiveGraphExecutor(store, planner, {}) + + # Run initial + report = await executor.run("run_002") + assert "approval_task" in report.waiting + + # Record human rejection + store.record_approval_decision("run_002", "approval_task", approved=False, payload={"reason": "security restriction"}) + assert store.node_state("run_002", "approval_task") == NodeState.CANCELLED + + # Check journal has task_cancelled event + events = [e.kind for e in store.events("run_002")] + assert "human_approval_denied" in events + assert "task_cancelled" in events + + +@pytest.mark.asyncio +async def test_transient_error_retry(tmp_path): + store = GraphStore(tmp_path / "test_retry.db") + initial_patch = GraphPatch(add=(TaskSpec("flaky_task", "flaky_skill"),)) + planner = SimplePlanner(initial_patch) + attempts = 0 + + async def flaky_skill(task: TaskSpec): + nonlocal attempts + attempts += 1 + if attempts == 1: + raise RuntimeError("RateLimitError: 429 Too Many Requests") + return {"result": "success on retry"} + + async def final_skill(task: TaskSpec): + return {"status": "done"} + + skills = { + "flaky_skill": flaky_skill, + "final_skill": final_skill, + } + policy = RetryPolicy(max_retries=3) + executor = LiveGraphExecutor(store, planner, skills, retry_policy=policy) + + report = await executor.run("run_003") + assert report.finished + assert attempts == 2 + assert store.node_state("run_003", "flaky_task") == NodeState.SUCCEEDED + assert store.node_state("run_003", "final_task") == NodeState.SUCCEEDED + + +@pytest.mark.asyncio +async def test_permanent_error_no_retry(tmp_path): + store = GraphStore(tmp_path / "test_permanent.db") + initial_patch = GraphPatch(add=(TaskSpec("perm_task", "perm_skill"),)) + planner = SimplePlanner(initial_patch) + attempts = 0 + + async def perm_skill(task: TaskSpec): + nonlocal attempts + attempts += 1 + raise PermissionError("PermissionDenied: Invalid tenant credential") + + skills = {"perm_skill": perm_skill} + policy = RetryPolicy(max_retries=3) + executor = LiveGraphExecutor(store, planner, skills, retry_policy=policy) + + await executor.run("run_004") + assert attempts == 1 + assert store.node_state("run_004", "perm_task") == NodeState.FAILED + + +@pytest.mark.asyncio +async def test_adversarial_approval_on_finished_graph(tmp_path): + """Adversarial Attack: Attempting to record approval on a non-waiting node or finished graph.""" + store = GraphStore(tmp_path / "test_adv.db") + initial_patch = GraphPatch( + add=(TaskSpec("task_1", "skill_1"),), + finish=True, + ) + planner = SimplePlanner(initial_patch) + + async def skill_1(task: TaskSpec): + return {"ok": True} + + skills = {"skill_1": skill_1} + executor = LiveGraphExecutor(store, planner, skills) + + report = await executor.run("run_adv") + assert report.finished + + # Adversarial approval call on a non-waiting task / finished graph + with pytest.raises((GraphMutationError, KeyError)): + store.record_approval_decision("run_adv", "task_1", approved=True) +