feat(live-graph): add human approval wait state and transient retry p…

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`.
+

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",
 ]
+
+

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)

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

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)))

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())