feat: restrict primary worker agent to enforce delegation (#6)

* feat: restrict primary worker agent to enforce delegation

* feat: removed read_file access from primary

* chore: corrected stale plugin description

* feat: switched to blocking delegate only tool calls

* feat: allow primary agent to validate subagent work

* feat: guide subagent to report consise findings

* feat: allow inter-tool narration for subagent

* feat: tidied up progress context

* feat: strengthened primary agents reading delegation prompt

* feat: improved semantic structure of agent prompts

* feat: improved semantic structure of staste persistence

* feat: order live stream events by timestamp

Author: jahilldev

api/.hermes/config.base.yaml

@@ -13,6 +13,7 @@ plugins:
     - observability/langfuse
     - capture_reasoning # backfills reasoning_content/<think> into the langfuse span so thinking renders in the stream
     - synchronous_delegate # forces delegate_task to run synchronously so the real result returns in-band (captured to PROGRESS.md) and nothing is lost
+    - worker_guard # blocks write_file/patch for the IMPLEMENT/REVISE primary (subagents unaffected) so every edit is delegated; the primary may still read to verify
 
 browser:
   inactivity_timeout: 120

api/.hermes/plugins/persist_state/__init__.py

@@ -24,14 +24,15 @@
 
 import json
 import os
+import re as _re
 import threading
 from typing import Any
 
 _PROGRESS_REL = os.path.join(".olympian", "PROGRESS.md")
 _CHECKLIST_HEADER = "## Checklist"
 _FINDINGS_HEADER = "## Findings"
-_FINDINGS_BUDGET = 12_000  # max chars kept in Findings (drop oldest beyond this) — keep re-reads cheap
-_ENTRY_CAP = 2_000  # max chars of any single subagent report
+_FINDINGS_BUDGET = 50_000  # max chars kept in Findings (drop oldest beyond this)
+_ENTRY_CAP = 12_000  # max chars of any single subagent report (holds a thorough survey whole)
 
 _LOCK = threading.Lock()
 _STATE: dict[str, Any] = {
@@ -111,6 +112,10 @@ def _seed() -> None:
         checklist, findings = _split(_read())
         _STATE["checklist"] = checklist
         _STATE["findings"] = findings
+        # Continue Findings numbering from the highest existing entry so a retry (a fresh process
+        # that resumes this file) doesn't restart at 1 and produce a jumbled "1, 2, 1" sequence.
+        nums = [int(m) for m in _re.findall(r"(?m)^### (\d+)\.", findings or "")]
+        _STATE["delegations"] = max(nums) if nums else 0
 
 
 def _flush() -> None:
@@ -154,29 +159,120 @@ def _render_checklist(result: Any) -> str:
     return "\n".join(lines)
 
 
-def _summarise_delegation(result: Any) -> str:
-    def one(r: Any) -> str:
-        if isinstance(r, dict):
-            return str(
-                r.get("summary")
-                or r.get("error")
-                or r.get("result")
-                or json.dumps(r, ensure_ascii=False, default=str)
-            )
+_THINK_TAGS = _re.compile(r"<(antThinking|think|thinking|reasoning)>.*?</\\1>", _re.DOTALL | _re.IGNORECASE)
+
+
+def _strip_thinking(text: str) -> str:
+    return _THINK_TAGS.sub("", text).strip()
+
+
+_HEADING_RE = _re.compile(r"^[ \t]*(#{1,6})[ \t]+(.+?)[ \t]*#*$")
+
+
+def _relevel_headings(text: str, base: int = 4) -> str:
+    """Re-base the headings in captured (model-authored) goal/report text so the shallowest sits at
+    ``base`` — one level below a "### N." Findings entry (h3) — while preserving the content's own
+    relative hierarchy. This keeps the agent's structure intact AND nested correctly under the entry,
+    so its headings never sit above, or collide with, the structural ## Findings / ### N. markers
+    when the file is injected into a prompt. Headings inside fenced code blocks are left untouched.
+    """
+    lines = text.split("\n")
+
+    def heading_levels():
+        in_fence = False
+        for line in lines:
+            s = line.lstrip()
+            if s.startswith("```") or s.startswith("~~~"):
+                in_fence = not in_fence
+                continue
+            m = None if in_fence else _HEADING_RE.match(line)
+            if m:
+                yield len(m.group(1))
+
+    levels = list(heading_levels())
+    if not levels:
+        return text
+    shift = base - min(levels)
+
+    out: list[str] = []
+    in_fence = False
+    for line in lines:
+        s = line.lstrip()
+        if s.startswith("```") or s.startswith("~~~"):
+            in_fence = not in_fence
+            out.append(line)
+            continue
+        m = None if in_fence else _HEADING_RE.match(line)
+        if m:
+            level = max(1, min(6, len(m.group(1)) + shift))
+            out.append("#" * level + " " + m.group(2).strip())
+        else:
+            out.append(line)
+    return "\n".join(out)
+
+
+_DONE = {"completed", "complete", "success", "succeeded", "ok"}
+
+
+def _one_result(r: Any) -> str:
+    """Render a single child task result. Only a COMPLETED child contributes its summary; an
+    incomplete one (timeout / max_iterations / error / interrupted) is recorded as a short marker
+    instead of its truncated mid-thought narration — capturing that pollutes Findings."""
+    if not isinstance(r, dict):
         return str(r)
+    status = str(r.get("status") or "").strip().lower()
+    exit_reason = str(r.get("exit_reason") or "").strip().lower()
+    # No status fields at all (older/other shapes) → treat as a plain result.
+    if (status or exit_reason) and status not in _DONE and exit_reason not in _DONE:
+        why = status or exit_reason or "unknown"
+        detail = str(r.get("error") or "").strip() or "cut off before producing a final report"
+        return f"⚠️ Sub-agent did not finish (status: {why}) — {detail[:200]}"
+    return str(r.get("summary") or r.get("result") or r.get("error") or "")
+
+
+def _summarise_delegation(result: Any) -> str:
+    one = _one_result
 
     if isinstance(result, str):
         try:
             result = json.loads(result)
         except Exception:
-            return result.strip()[:_ENTRY_CAP]
-    text = "\n".join(one(r) for r in result) if isinstance(result, list) else one(result)
-    text = text.strip()
+            return _strip_thinking(result)[:_ENTRY_CAP]
+
+    # delegate_task returns {"results": [ {summary,...}, ... ], "note": ...}; older/other shapes
+    # may be a bare list or dict.
+    if isinstance(result, dict) and isinstance(result.get("results"), list):
+        items = result["results"]
+    elif isinstance(result, list):
+        items = result
+    else:
+        items = [result]
+
+    text = "\n".join(p for p in (one(r) for r in items) if p.strip())
+    text = _strip_thinking(text)
+    text = _relevel_headings(text)
     if len(text) > _ENTRY_CAP:
         text = text[:_ENTRY_CAP].rstrip() + " …[trimmed]"
     return text
 
 
+def _heading_label(goal: Any, cap: int = 100) -> str:
+    """A short one-line label for a Findings heading. The primary writes the whole task as a single
+    long line, so we take the first line, cut it at the first sentence/clause boundary if that is
+    short enough, then hard-cap on a word boundary."""
+    line = str(goal or "").strip().splitlines()
+    line = line[0].strip() if line else ""
+    # Cut at the first clause/sentence boundary that is actually present and short enough.
+    for sep in (": ", ". "):
+        if sep in line:
+            head = line.split(sep, 1)[0]
+            if 0 < len(head) <= cap:
+                return head
+    if len(line) > cap:
+        line = line[:cap].rsplit(" ", 1)[0].rstrip(" ,.;:") + "…"
+    return line
+
+
 def _is_dispatch_ack(result):
     """A background delegation's post-hook result is only a dispatch acknowledgement, not the
     subagent's report (that arrives asynchronously) — recognise it so we don't record it."""
@@ -254,11 +350,14 @@ def on_post_tool_call(
                 if not _is_dispatch_ack(result):
                     _STATE["delegations"] += 1
                     n = _STATE["delegations"]
-                    goal = ""
-                    if isinstance(args, dict):
-                        first = str(args.get("goal", "")).strip().splitlines()
-                        goal = first[0] if first else ""
-                    entry = (f"### {n}. {goal}".rstrip()) + "\n" + _summarise_delegation(result) + "\n"
+                    goal_full = str(args.get("goal", "")).strip() if isinstance(args, dict) else ""
+                    label = _heading_label(goal_full)
+                    # Short heading for readability; the full goal is kept as body so no context is
+                    # lost, then the subagent's report.
+                    entry = f"### {n}. {label}".rstrip() + "\n"
+                    if goal_full and goal_full != label:
+                        entry += f"**Goal:** {_relevel_headings(goal_full)}\n\n"
+                    entry += _summarise_delegation(result) + "\n"
                     prior = _STATE["findings"] or ""
                     _STATE["findings"] = (prior + "\n\n" + entry) if prior else entry
                     _flush()

api/.hermes/plugins/worker_guard/__init__.py

@@ -0,0 +1,152 @@
+"""worker_guard — force the IMPLEMENT/REVISE primary to delegate every edit.
+
+The primary may READ files (to verify a sub-agent's work), but it must not EDIT them: every change
+goes through a sub-agent so the work is delegated, captured in PROGRESS.md's Findings, and the
+primary's context stays manageable. Two block paths, both primary-only:
+
+  * the write tools — write_file / patch
+  * file-mutating shell commands — `sed -i`, output redirects (> / >>), tee, mv/cp/rm/…, so the
+    primary can't bypass the tool block via terminal. Test/build commands and harmless redirects
+    (2>&1, > /dev/null) are left alone.
+
+We cannot remove the write tools from the parent's toolset — Hermes caps a child's toolset to a
+subset of the parent's (see delegate_tool's intersection), so that would strip editing from
+sub-agents too. Instead we keep the tools and BLOCK them for the primary only, via a pre_tool_call
+directive. The prompt tells the model they're disabled so it delegates up front.
+
+Primary vs sub-agent is decided by task_id without relying on call ordering: the first tool call in
+the process is the parent's (a child only exists after a delegate_task), and any task_id seen WHILE a
+delegation is in flight is a sub-agent's. The hooks never raise — a guard failure must not break a
+run (it just declines to block).
+"""
+from __future__ import annotations
+
+import os
+import re
+import shlex
+import threading
+from typing import Any, Optional
+
+_WORKER_PHASES = {"IMPLEMENT", "REVISE"}
+# Write tools — the primary may still read (read_file/search_files/cat) to verify a sub-agent.
+_BLOCKED = {"write_file", "patch"}
+
+# Shell commands that mutate the filesystem (checked per pipeline segment, ignoring a leading `cd`).
+_WRITE_CMDS = {
+    "tee", "mv", "cp", "rm", "rmdir", "touch", "mkdir", "dd", "truncate", "install", "ln",
+    "patch", "chmod", "chown",
+}
+# A redirect operator with its target: optional leading fd (or &), > or >>, then the target token.
+_REDIR = re.compile(r"(?:\d*|&)\s*>>?\s*(\S+)")
+_SEGMENTS = re.compile(r"\|\||&&|;|\|")
+
+_LOCK = threading.Lock()
+_STATE: dict[str, Any] = {
+    "primary": None,  # the primary agent's task_id, once known
+    "active": 0,  # in-flight delegate_task calls
+    "subagents": set(),  # task_ids identified as delegated sub-agents
+}
+
+
+def _phase_ok() -> bool:
+    return (os.environ.get("OLYMPIAN_PHASE") or "").strip().upper() in _WORKER_PHASES
+
+
+def _has_write_redirect(cmd: str) -> bool:
+    for m in _REDIR.finditer(cmd):
+        target = m.group(1).strip("'\"")
+        if target == "/dev/null" or re.fullmatch(r"&?\d+", target):
+            continue  # > /dev/null or an fd dup (2>&1, >&2) — not a file write
+        return True
+    return False
+
+
+def _has_write_command(cmd: str) -> bool:
+    for seg in _SEGMENTS.split(cmd):
+        seg = seg.strip()
+        if not seg or seg.startswith("cd "):
+            continue
+        try:
+            parts = shlex.split(seg)
+        except Exception:
+            parts = seg.split()
+        if not parts:
+            continue
+        word = parts[0]
+        if word in _WRITE_CMDS:
+            return True
+        if word == "sed" and any(p == "-i" or p.startswith("-i") for p in parts[1:]):
+            return True  # in-place edit; plain sed is a read/filter
+    return False
+
+
+def _is_file_write_command(cmd: Optional[str]) -> bool:
+    cmd = (cmd or "").strip()
+    if not cmd:
+        return False
+    return _has_write_redirect(cmd) or _has_write_command(cmd)
+
+
+def on_pre_tool_call(
+    *, tool_name: str = "", task_id: str = "", args: Optional[dict] = None, **_: Any
+) -> Any:
+    if not _phase_ok():
+        return None
+    try:
+        with _LOCK:
+            if tool_name == "delegate_task":
+                if _STATE["primary"] is None and task_id:
+                    _STATE["primary"] = task_id  # only the primary delegates
+                _STATE["active"] += 1
+                return None
+
+            # A tool call during an in-flight delegation, under a non-primary id, is a sub-agent's.
+            if _STATE["active"] > 0 and task_id and task_id != _STATE["primary"]:
+                _STATE["subagents"].add(task_id)
+            if _STATE["primary"] is None and task_id:
+                _STATE["primary"] = task_id  # first non-delegate tool call = the parent
+            is_subagent = task_id in _STATE["subagents"]
+
+        if is_subagent:
+            return None  # sub-agents do the editing — never block them
+
+        if tool_name in _BLOCKED:
+            return {
+                "action": "block",
+                "message": (
+                    f"`{tool_name}` is disabled for you (the orchestrator) — you must not edit files "
+                    "yourself. Delegate this change to a sub-agent with delegate_task; it edits in its "
+                    "own context and returns the result to you. (You can still read files directly to "
+                    "verify a sub-agent's work.)"
+                ),
+            }
+
+        if tool_name == "terminal" and _is_file_write_command((args or {}).get("command")):
+            return {
+                "action": "block",
+                "message": (
+                    "You must not modify files through the shell (no in-place sed, > / >> redirects, "
+                    "tee, mv/cp/rm, etc.) — terminal is for running tests/builds only. Delegate this "
+                    "change to a sub-agent with delegate_task instead."
+                ),
+            }
+    except Exception:
+        return None
+    return None
+
+
+def on_post_tool_call(*, tool_name: str = "", **_: Any) -> Any:
+    if not _phase_ok():
+        return None
+    try:
+        if tool_name == "delegate_task":
+            with _LOCK:
+                _STATE["active"] = max(0, _STATE["active"] - 1)
+    except Exception:
+        pass
+    return None
+
+
+def register(ctx) -> None:
+    ctx.register_hook("pre_tool_call", on_pre_tool_call)
+    ctx.register_hook("post_tool_call", on_post_tool_call)

api/.hermes/plugins/worker_guard/plugin.yaml

@@ -0,0 +1,6 @@
+name: worker_guard
+version: 1.0.0
+description: "Blocks write_file/patch AND file-mutating shell commands (sed -i, > / >> redirects, tee, mv/cp/rm, ...) for the IMPLEMENT/REVISE PRIMARY agent (sub-agents unaffected) so every edit is delegated. The primary may still read files (read_file/search_files/cat) and run tests/builds via terminal to verify a sub-agent's work. We block rather than remove the write tools because Hermes caps a child's toolset to a subset of the parent's, so removing them from the parent would also strip editing from sub-agents."
+hooks:
+  - pre_tool_call
+  - post_tool_call