Add TODOS checklist and DELEGATE context-isolated subtask features

.env.example

@@ -139,6 +139,11 @@ CLK_ROBUSTNESS_REFINE_ACCEPT_THRESHOLD=0.8
 CLK_ROBUSTNESS_QA_PARALLEL_JUDGES=1
 CLK_ROBUSTNESS_MAX_QA_DEPTH=3
 
+# Context-isolated DELEGATE sub-agents (DELEGATE: TO: <agent> TASK: ...).
+# max_delegate_depth caps nesting; 1 = a worker may delegate but the child
+# cannot itself delegate (one level deep).
+CLK_ROBUSTNESS_MAX_DELEGATE_DEPTH=1
+
 # Ralph / autoresearch plateau detection. After plateau_window
 # consecutive iterations without improvement, the loop escalates +
 # reframes; failing that, it terminates gracefully with done.md.

.gitignore

@@ -41,6 +41,12 @@ Thumbs.db
 # Kickoff workspace (kickoff dirs created by kickoff.sh land here)
 workspace/
 
+# Context-offload scratch space: agents park large command output / generated
+# data / logs here and read back only slices (grep/head/sed). Disposable
+# working memory, never a deliverable — kept out of git so it can't pollute
+# history via the per-run `git add -A` auto-commit.
+scratch/
+
 # Local env overrides (kickoff.sh reads .env if present)
 .env
 

clk_harness/config.py

@@ -261,6 +261,10 @@ def save_json(path: Path, data: Dict[str, Any], *, backup: bool = True) -> None:
         # Inter-agent Q&A bounds.
         "qa_parallel_judges": 1,
         "max_qa_depth": 6,
+        # Context-isolated DELEGATE sub-agents. Depth 1 = a worker may spawn an
+        # isolated child for a bounded subtask, but that child cannot itself
+        # delegate (one level deep) — bounds token cost and recursion.
+        "max_delegate_depth": 1,
         # Ralph / autoresearch plateau detection.
         # A large window means the team runs many more iterations before the
         # harness decides there is no more improvement to extract.

clk_harness/kickoff.py

@@ -106,6 +106,7 @@ def _csv(s) -> List[str]:
     ("CLK_ROBUSTNESS_REFINE_ACCEPT_THRESHOLD", ("robustness", "refine_accept_threshold"), float),
     ("CLK_ROBUSTNESS_QA_PARALLEL_JUDGES", ("robustness", "qa_parallel_judges"), int),
     ("CLK_ROBUSTNESS_MAX_QA_DEPTH", ("robustness", "max_qa_depth"), int),
+    ("CLK_ROBUSTNESS_MAX_DELEGATE_DEPTH", ("robustness", "max_delegate_depth"), int),
     ("CLK_ROBUSTNESS_PLATEAU_WINDOW", ("robustness", "plateau_window"), int),
     ("CLK_ROBUSTNESS_PLATEAU_ACTION", ("robustness", "plateau_action"), str),
     ("CLK_ROBUSTNESS_DEBATE", ("robustness", "debate"), str),

clk_harness/orchestration/agent/prompts.py

@@ -17,6 +17,7 @@
 from ...utils.activity_log import log_event
 from .. import blackboard as _blackboard
 from .. import casting as _casting
+from .. import todos as _todos
 
 if TYPE_CHECKING:
     import threading
@@ -94,7 +95,11 @@ def _observer_log(self, line: str) -> None: ...
     def render_prompt(self, agent: "AgentSpec", objective: str, extra: Optional[Dict[str, Any]] = None) -> str:
         try:
             template = self._load_prompt_template(agent.prompt_file)
-            ctx = self._collect_context(objective, extra or {})
+            # Thread the dispatched agent's name into context so the per-author
+            # $todos checklist (and $agent) resolve on the normal dispatch path,
+            # where ``extra`` would otherwise carry no agent name.
+            merged = {**(extra or {}), "agent": (extra or {}).get("agent") or agent.name}
+            ctx = self._collect_context(objective, merged)
             return self._safe_substitute(template, ctx)
         except Exception as exc:
             log_exception("orchestration.agent.render_prompt", exc)
@@ -407,22 +412,33 @@ def _collect_context(self, objective: str, extra: Dict[str, Any]) -> Dict[str, A
         # provided, otherwise show the most recent global posts so the
         # agent has at least some peer context. Capped to keep prompts
         # bounded; tunable via clk.config.json::blackboard.
-        bb_cfg = (self.clk_cfg.get("blackboard") or {})
-        bb_inputs = list(extra.get("blackboard_inputs") or [])
-        # Allow the chief to widen the digest via stage metadata flag
-        # ``include_full_blackboard`` (carried through ``extra``).
-        if extra.get("include_full_blackboard"):
-            bb_inputs = []
-        try:
-            bb_digest = _blackboard.digest(
-                self.paths,
-                selectors=bb_inputs,
-                max_posts=int(bb_cfg.get("digest_max_posts") or 20),
-                max_chars_per_post=int(bb_cfg.get("digest_max_chars_per_post") or 800),
+        # Context isolation for DELEGATE children: a delegated subtask must NOT
+        # inherit the caller's blackboard. Note an empty selector list does
+        # NOT isolate (digest() returns recent global posts when selectors are
+        # falsy), so this needs its own explicit branch.
+        if extra.get("delegate_isolated") or str(extra.get("phase") or "") == "delegate":
+            bb_digest = (
+                "Blackboard digest: (isolated — this is a delegated subtask; "
+                "peer context is intentionally withheld. Work only from the "
+                "task described in your objective.)"
             )
-        except Exception as exc:
-            log_exception("orchestration.agent._collect_context.blackboard", exc)
-            bb_digest = "Blackboard digest: (unavailable)"
+        else:
+            bb_cfg = (self.clk_cfg.get("blackboard") or {})
+            bb_inputs = list(extra.get("blackboard_inputs") or [])
+            # Allow the chief to widen the digest via stage metadata flag
+            # ``include_full_blackboard`` (carried through ``extra``).
+            if extra.get("include_full_blackboard"):
+                bb_inputs = []
+            try:
+                bb_digest = _blackboard.digest(
+                    self.paths,
+                    selectors=bb_inputs,
+                    max_posts=int(bb_cfg.get("digest_max_posts") or 20),
+                    max_chars_per_post=int(bb_cfg.get("digest_max_chars_per_post") or 800),
+                )
+            except Exception as exc:
+                log_exception("orchestration.agent._collect_context.blackboard", exc)
+                bb_digest = "Blackboard digest: (unavailable)"
 
         # Casting-rejection feedback: surface duplicate-prevention misses
         # so the chief learns from them on the next dispatch.
@@ -444,6 +460,17 @@ def _collect_context(self, objective: str, extra: Dict[str, Any]) -> Dict[str, A
             except Exception as exc:
                 log_exception("orchestration.agent._collect_context.notes", exc)
 
+        # Working checklist: inject THIS author's own mutable TODOS list so it
+        # can review and re-emit an updated checklist this turn. Per-author, so
+        # peers' lists stay out of the way (that is what the blackboard is for).
+        try:
+            todos = _todos.render_todos(
+                _todos.todos_for(self.paths, str(extra.get("agent") or ""))
+            )
+        except Exception as exc:
+            log_exception("orchestration.agent._collect_context.todos", exc)
+            todos = "(no todos yet)"
+
         # Outputs contract: convert the stage_outputs list into a concrete,
         # agent-visible instruction block so workers know BEFORE they write
         # their first response which POST PRODUCES keys are required. Without
@@ -483,6 +510,7 @@ def _collect_context(self, objective: str, extra: Dict[str, Any]) -> Dict[str, A
             "blackboard_digest": bb_digest,
             "casting_feedback": casting_feedback or "(none)",
             "notes": notes,
+            "todos": todos,
             "outputs_contract": outputs_contract,
         }
         # ``telemetry`` is a live counter object threaded through ``extra`` for

clk_harness/orchestration/agent/runner.py

@@ -52,6 +52,9 @@ def __init__(
         # Lock around meta-prompt cache reads/writes so parallel stages
         # racing to draft the same dispatch prompt don't corrupt the file.
         self._meta_cache_lock = threading.Lock()
+        # Serialises the mutable per-author TODOS store (.clk/state/todos.json)
+        # so parallel dispatches sharing one agent name can't tear the write.
+        self._todos_lock = threading.Lock()
 
     # -- public ------------------------------------------------------------
 
@@ -103,6 +106,9 @@ def get_provider(self, name: Optional[str]) -> AgentProvider:
         "qa_answer",
         "refine_critic",
         "refine_worker",
+        # A context-isolated DELEGATE child: single-shot, must not recurse
+        # into consensus / quality / QA / further delegation.
+        "delegate",
         # Mission-level chief dispatches: single-shot planning / gating that
         # must not recurse into consensus or the quality-retry loop.
         "charter",
@@ -631,6 +637,10 @@ def _on_progress(kind: str, message: str) -> None:
         # apply hooks. Posting is cheap and uncommitted, so it happens
         # even for dry-runs to keep the digest accurate during planning.
         self._apply_posts(run, extra or {})
+        # Persist any TODOS: checklist block. Mutable per-author working
+        # memory; like posts it runs even for dry-runs so the checklist
+        # stays live during planning.
+        self._apply_todos(run, extra or {})
         # Apply any PROPOSE_ROLE / PROPOSE_WORKFLOW blocks the agent
         # emitted. Mutates ``self.agents_cfg`` in place so the very next
         # stage that names a freshly-proposed role can dispatch to it.
@@ -643,6 +653,9 @@ def _on_progress(kind: str, message: str) -> None:
         # files_written list so the TUI / commit logic see them.
         if not is_dry:
             self._apply_actions(run, extra or {})
+            # Spawn any DELEGATE children AFTER the parent's own actions/commit
+            # land, so the parent's work is a distinct unit from the child's.
+            self._apply_delegate(run, extra or {})
         if self.observer is not None:
             try:
                 self.observer.end(agent.name, run)
@@ -689,7 +702,9 @@ def _apply_consensus(self, run: AgentRun, extra: Dict[str, Any]) -> None:
         text = run.response.text or ""
         if not text or "PROPOSE_CONSENSUS" not in text:
             return
-        if str(extra.get("phase") or "") == "consensus":
+        # Don't fan out consensus from a consensus sample (recursion) or from a
+        # context-isolated DELEGATE child (it must stay single-shot).
+        if str(extra.get("phase") or "") in {"consensus", "delegate"}:
             return
         proposals = _casting.parse_consensus_proposals(text)
         if not proposals:

clk_harness/orchestration/agent/transcript.py

@@ -10,6 +10,7 @@
 
 import json
 from dataclasses import dataclass, field
+from datetime import datetime
 from typing import TYPE_CHECKING, Any, Dict, FrozenSet, List, Optional
 
 from ...git_ops import add_all, has_changes, head_sha, is_repo
@@ -20,6 +21,7 @@
 from .. import actions as _actions
 from .. import blackboard as _blackboard
 from .. import casting as _casting
+from .. import todos as _todos
 
 if TYPE_CHECKING:
     import threading
@@ -125,6 +127,7 @@ class TranscriptMixin:
     clk_cfg: Dict[str, Any]
     observer: Optional[AgentObserver]
     _proposals_lock: "threading.RLock"
+    _todos_lock: "threading.Lock"
     _META_PHASES: FrozenSet[str]
 
     if TYPE_CHECKING:
@@ -184,7 +187,14 @@ def _apply_posts(self, run: AgentRun, extra: Dict[str, Any]) -> None:
         That makes the asker's worker effectively block on the answer,
         which gets posted back to the blackboard with ``post_type:
         answer`` and ``consumes: [<question_id>]``.
+
+        A context-isolated DELEGATE child runs with ``suppress_posts`` (and
+        ``phase == "delegate"``): its own POST blocks are NOT persisted to the
+        shared board — only its distilled result returns to the caller as a
+        single ``delegate_result`` post created by ``_apply_delegate``.
         """
+        if str(extra.get("phase") or "") == "delegate" or extra.get("suppress_posts"):
+            return
         text = (run.response.text or "")
         if not text or "POST:" not in text:
             return
@@ -312,6 +322,158 @@ def _route_blocking_questions(
                 dry_run=self.clk_cfg.get("dry_run", False),
             )
 
+    def _apply_todos(self, run: AgentRun, extra: Dict[str, Any]) -> None:
+        """Persist a ``TODOS:`` block the agent emitted as its live checklist.
+
+        The checklist is mutable and per-author: the latest block overwrites
+        this author's previous list (last-write-wins), and is re-injected into
+        this author's next prompt via the ``$todos`` placeholder. Skipped for
+        meta phases (consensus / qa_answer / delegate / etc.) so a subtask's
+        stray block can't clobber the driving agent's checklist.
+        """
+        if str(extra.get("phase") or "") in self._META_PHASES:
+            return
+        text = run.response.text or ""
+        if "TODOS:" not in text:
+            return
+        try:
+            with self._todos_lock:
+                _todos.apply_todos_blocks(
+                    self.paths,
+                    text,
+                    author=run.agent,
+                    stage_id=str(extra.get("stage_id") or ""),
+                    workflow=str(extra.get("workflow") or ""),
+                )
+        except Exception as exc:
+            log_exception("orchestration.agent._apply_todos", exc)
+
+    def _apply_delegate(self, run: AgentRun, extra: Dict[str, Any]) -> None:
+        """Spawn context-isolated DELEGATE children for a bounded subtask.
+
+        Each ``DELEGATE:`` block dispatches the named target once, with the
+        caller's blackboard withheld (``delegate_isolated``) and the child's
+        own POST blocks suppressed. The child MAY do real work — its ACTION
+        blocks execute and commit under its own name. The child's distilled
+        result returns to the caller as a single ``delegate_result`` post it
+        sees on its next turn.
+
+        Skipped from any meta phase (so a child cannot itself delegate) and
+        bounded by ``max_delegate_depth`` (default 1), a per-turn cap, and
+        unknown-target / self / cycle guards — mirroring the blocking-Q&A
+        routing in ``_route_blocking_questions``.
+        """
+        text = run.response.text or ""
+        if "DELEGATE:" not in text:
+            return
+        if str(extra.get("phase") or "") in self._META_PHASES:
+            return
+        props = _casting.parse_delegate_proposals(text)
+        if not props:
+            return
+        cfg = self.clk_cfg.get("robustness") or {}
+        max_depth = int(cfg.get("max_delegate_depth") or 1)
+        chain: List[str] = list(extra.get("delegate_chain") or [])
+        if len(chain) >= max_depth:
+            log_event(
+                self.paths,
+                "delegate_chain_capped",
+                agent=run.agent,
+                depth=len(chain),
+                max_depth=max_depth,
+                chain=list(chain),
+            )
+            return
+        max_per_turn = int(cfg.get("max_delegates_per_turn") or 2)
+        result_cap = int(cfg.get("delegate_result_max_chars") or 2000)
+        known = set((self.agents_cfg.get("agents") or {}).keys())
+        next_chain = chain + [run.agent]
+        for prop in props[:max_per_turn]:
+            target = prop.target
+            if not target or target not in known:
+                log_event(
+                    self.paths,
+                    "delegate_target_unknown",
+                    agent=run.agent,
+                    target=target,
+                    name=prop.name,
+                )
+                continue
+            if target == run.agent or target in chain:
+                log_event(
+                    self.paths,
+                    "delegate_chain_cycle",
+                    agent=run.agent,
+                    target=target,
+                    chain=list(chain),
+                )
+                continue
+            req_id = (
+                f"deleg-{run.agent}-"
+                f"{datetime.now().strftime('%Y%m%dT%H%M%S%f')}-{prop.name}"
+            )
+            child_obj = (
+                "Delegated, context-isolated subtask.\n\n"
+                f"Requested by `{run.agent}`. You do NOT see the caller's "
+                "blackboard — work only from what is written here. You MAY do "
+                "real work (emit ACTION blocks to change files); they will be "
+                "committed under your name. When done, end with a concise, "
+                "self-contained summary of the result the caller needs — that "
+                "summary is all that is returned to them.\n"
+            )
+            if prop.context:
+                child_obj += f"\nContext:\n{prop.context}\n"
+            child_obj += f"\nTask:\n{prop.objective}\n"
+            log_event(
+                self.paths,
+                "delegate_dispatch",
+                agent=run.agent,
+                target=target,
+                name=prop.name,
+                req_id=req_id,
+                chain=next_chain,
+            )
+            self._observer_log(f"delegate :: {run.agent} → {target} :: {prop.name}")
+            try:
+                child = self._dispatch_once(
+                    target,
+                    child_obj,
+                    extra={
+                        "phase": "delegate",
+                        "delegate_chain": next_chain,
+                        "delegate_isolated": True,
+                        "suppress_posts": True,
+                        "agent": target,
+                        "stage_id": extra.get("stage_id"),
+                        "workflow": extra.get("workflow"),
+                    },
+                    dry_run=self.clk_cfg.get("dry_run", False),
+                )
+            except Exception as exc:
+                log_exception("orchestration.agent._apply_delegate.dispatch", exc)
+                continue
+            body = ""
+            if child is not None and child.response is not None:
+                body = (child.response.text or "").strip()
+            if len(body) > result_cap:
+                body = body[:result_cap].rstrip() + " …"
+            try:
+                p = _blackboard.post(
+                    self.paths,
+                    author=run.agent,
+                    body=body or "(delegate produced no output)",
+                    post_type="delegate_result",
+                    consumes=[req_id],
+                    produces=[f"delegate:{prop.name}"],
+                    stage_id=str(extra.get("stage_id") or ""),
+                    workflow=str(extra.get("workflow") or ""),
+                    slug_hint=f"delegate-{prop.name}",
+                )
+                if p.id and p.id not in run.posts:
+                    run.posts.append(p.id)
+            except Exception as exc:
+                log_exception("orchestration.agent._apply_delegate.post", exc)
+
     def _apply_actions(self, run: AgentRun, extra: Optional[Dict[str, Any]] = None) -> None:
         """Execute ACTION blocks; merge harness-written files back into the run."""
         extra = extra or {}