diff --git a/engine/hooks/frustration-watchdog/claude_stop_check.py b/engine/hooks/frustration-watchdog/claude_stop_check.py
index 89a01493f..8b9e6f513 100755
--- a/engine/hooks/frustration-watchdog/claude_stop_check.py
+++ b/engine/hooks/frustration-watchdog/claude_stop_check.py
@@ -34,8 +34,15 @@
 import os
 import re
 import sys
+import hashlib
 from datetime import datetime
 
+sys.path.insert(0, os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk"))
+
+from finding import Finding  # noqa: E402
+from runtime import run_hook  # noqa: E402
+
 IMPATIENCE_PATTERNS = [
     ("profanity", re.compile(r"\b(fuck\w*|wtf|shit\w*|goddamn|dammit|damn it|stupid)\b", re.I)),
     ("told-you", re.compile(r"\bi (already |just )?told you\b|\bi asked you not\b|\bi already said\b", re.I)),
@@ -83,6 +90,8 @@
 """How Claude Code records a tool call a PreToolUse hook refused: an is_error
 tool_result whose text opens "PreToolUse:Bash hook error: [<command>]: ..."."""
 
+RULE_PREFIX = "frustration-watchdog"
+
 
 def _is_allcaps(text):
     letters = [c for c in text if c.isalpha()]
@@ -203,27 +212,23 @@ def ends_the_wait(message):
     return bool(NEXT_STEP_RE.search(message) or ETA_RE.search(message))
 
 
-def main():
-    try:
-        data = json.load(sys.stdin)
-    except json.JSONDecodeError:
-        return
+def detect(data):
     if data.get("stop_hook_active") or data.get("agent_id"):
-        return
+        return []
     message = data.get("last_assistant_message") or ""
     transcript_path = data.get("transcript_path") or ""
     if not message or not transcript_path or not os.path.isfile(transcript_path):
-        return
+        return []
     try:
         msgs = human_user_messages(transcript_path)
         kinds = impatience_kinds(msgs)
     except Exception as exc:
         print(f"catstack-hook-error frustration-watchdog: {type(exc).__name__}: {exc}", file=sys.stderr)
-        return  # fail open: a broken watchdog must never brick a session
+        return []
     if not kinds:
-        return
+        return []
     if ends_the_wait(message):
-        return
+        return []
     try:
         refused = turn_has_hook_refusal(transcript_path)
         unchecked = None
@@ -236,26 +241,48 @@ def main():
         "and this reply hands them nothing visible. "
     )
     if refused:
-        sys.stderr.write(
-            head + "A hook refused a tool call this turn, so you are the one blocked: "
+        feedback = (
+            head
+            + "A hook refused a tool call this turn, so you are the one blocked: "
             "do not hand the user steps to work around it. End the wait: ask them a "
             "direct question, or state an explicit no-action window "
-            "(\"nothing needed from you for ~2 min\"). Per CLAUDE.md live-demo rules.\n"
+            "(\"nothing needed from you for ~2 min\"). Per CLAUDE.md live-demo rules."
         )
     else:
-        sys.stderr.write(
-            head + "End the wait: give exactly one "
+        feedback = (
+            head
+            + "End the wait: give exactly one "
             "concrete action for the user (\"click X\", \"run Y\", \"say Z\"), ask them a "
             "direct question, or state an explicit no-action window "
-            "(\"nothing needed from you for ~2 min\"). Per CLAUDE.md live-demo rules.\n"
+            "(\"nothing needed from you for ~2 min\"). Per CLAUDE.md live-demo rules."
         )
     if unchecked:
-        sys.stderr.write(
-            f"(frustration-watchdog could not read this turn's tool results ({unchecked}), "
+        feedback = (
+            f"catstack-hook-error frustration-watchdog: {unchecked}\n"
+            + feedback
+            + "\n"
+            + f"(frustration-watchdog could not read this turn's tool results ({unchecked}), "
             "so it could not tell whether a hook refused a tool call; the wording above "
-            "is the default.)\n"
+            "is the default.)"
         )
-    sys.exit(2)
+    primary_kind = sorted(set(kinds))[0]
+    return [
+        Finding(
+            rule_id=f"{RULE_PREFIX}.{primary_kind}",
+            subject="reply:" + hashlib.sha256(message.encode("utf-8")).hexdigest(),
+            message=feedback,
+            evidence=", ".join(sorted(set(kinds))),
+        )
+    ]
+
+
+def main():
+    try:
+        run_hook("frustration-watchdog", "claude", detect, "Stop")
+    except SystemExit as exc:
+        if exc.code == 0:
+            return
+        raise
 
 
 if __name__ == "__main__":
diff --git a/engine/hooks/frustration-watchdog/codex_stop_check.py b/engine/hooks/frustration-watchdog/codex_stop_check.py
new file mode 100644
index 000000000..7d941052c
--- /dev/null
+++ b/engine/hooks/frustration-watchdog/codex_stop_check.py
@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+"""Codex Stop entrypoint for frustration-watchdog."""
+from __future__ import annotations
+
+import os
+import sys
+
+sys.path.insert(0, os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk"))
+
+from claude_stop_check import detect  # noqa: E402
+from runtime import run_hook  # noqa: E402
+
+
+def main() -> None:
+    run_hook("frustration-watchdog", "codex", detect, "Stop")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/engine/hooks/frustration-watchdog/cursor_stop_check.py b/engine/hooks/frustration-watchdog/cursor_stop_check.py
new file mode 100644
index 000000000..64986bfe1
--- /dev/null
+++ b/engine/hooks/frustration-watchdog/cursor_stop_check.py
@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+"""Cursor stop entrypoint for frustration-watchdog."""
+from __future__ import annotations
+
+import os
+import sys
+
+sys.path.insert(0, os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk"))
+
+from claude_stop_check import detect  # noqa: E402
+from runtime import run_hook  # noqa: E402
+
+
+def main() -> None:
+    run_hook("frustration-watchdog", "cursor", detect, "stop")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/engine/hooks/frustration-watchdog/tests/test_hooks_sdk_mode.py b/engine/hooks/frustration-watchdog/tests/test_hooks_sdk_mode.py
new file mode 100644
index 000000000..f26b8e619
--- /dev/null
+++ b/engine/hooks/frustration-watchdog/tests/test_hooks_sdk_mode.py
@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+from test_hooks import NARRATION, WAITING, human, transcript_lines, tool_turn
+
+HOOK_DIR = Path(__file__).resolve().parents[1]
+HOOK = HOOK_DIR / "claude_stop_check.py"
+
+
+def payload(transcript_path: str) -> dict[str, object]:
+    return {
+        "hook_event_name": "Stop",
+        "session_id": "frustration-watchdog-sdk-mode",
+        "transcript_path": transcript_path,
+        "last_assistant_message": NARRATION,
+        "stop_hook_active": False,
+    }
+
+
+def run_hook(data: dict[str, object], env: dict[str, str]) -> subprocess.CompletedProcess[str]:
+    merged_env = os.environ.copy()
+    merged_env.pop("CATSTACK_HOOK_MODE_FRUSTRATION_WATCHDOG", None)
+    merged_env.update(env)
+    return subprocess.run(
+        [sys.executable, str(HOOK)],
+        input=json.dumps(data),
+        capture_output=True,
+        text=True,
+        timeout=10,
+        env=merged_env,
+    )
+
+
+class SdkModeTest(unittest.TestCase):
+    def test_mode_override_warn_turns_stop_into_warning(self) -> None:
+        transcript = transcript_lines([human(WAITING)])
+        try:
+            result = run_hook(
+                payload(transcript),
+                {"CATSTACK_HOOK_MODE_FRUSTRATION_WATCHDOG": "warn"},
+            )
+        finally:
+            os.unlink(transcript)
+
+        self.assertEqual(0, result.returncode, result.stderr)
+        self.assertEqual("", result.stderr)
+        output = json.loads(result.stdout)
+        self.assertIn(
+            "The user's last message was impatience-shaped (waiting)",
+            output["hookSpecificOutput"]["additionalContext"],
+        )
+
+    def test_each_finding_writes_one_event_row_with_rule_id(self) -> None:
+        with tempfile.TemporaryDirectory() as metrics_dir:
+            transcript = transcript_lines([human(WAITING)] + tool_turn("Created PR #12", is_error=False))
+            try:
+                result = run_hook(
+                    payload(transcript),
+                    {
+                        "CATSTACK_HOOK_METRICS_DIR": metrics_dir,
+                        "CATSTACK_HOOK_MODE_FRUSTRATION_WATCHDOG": "warn",
+                    },
+                )
+            finally:
+                os.unlink(transcript)
+            rows = self._event_rows(metrics_dir)
+
+        self.assertEqual(0, result.returncode, result.stderr)
+        self.assertEqual(1, len(rows))
+        self.assertEqual("frustration-watchdog", rows[0]["hook"])
+        self.assertEqual("frustration-watchdog.waiting", rows[0]["rule_id"])
+        self.assertEqual("warn", rows[0]["mode"])
+        self.assertEqual("override", rows[0]["mode_source"])
+        self.assertEqual("warned", rows[0]["action"])
+
+    def _event_rows(self, metrics_dir: str) -> list[dict[str, object]]:
+        files = list(Path(metrics_dir).glob("events-*.jsonl"))
+        self.assertEqual(1, len(files))
+        return [json.loads(line) for line in files[0].read_text(encoding="utf-8").splitlines()]
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/product/skills/loop-generator/SKILL.md b/product/skills/loop-generator/SKILL.md
index 1043bf4c5..f7666ebbf 100644
--- a/product/skills/loop-generator/SKILL.md
+++ b/product/skills/loop-generator/SKILL.md
@@ -40,11 +40,13 @@ Collect and resolve every field below before drafting. Don't improvise a differe
 11. `fail_condition_rule` — repeated-attempt threshold and grouping key (e.g. "3 failures on the same (target, symptom) pair → stop and report, don't keep retrying").
 12. `local_proxy_command` — the safest repeatable local/proxy verification command, or `none`.
 13. `write_mode` — one of `diagnostic_only` (never mutate, just report), `worker_owned_writes` (the loop itself makes changes), or `choose_each_run` (ask each time).
+14. `state_artifact` — a file or digest an existing worker/cron/daemon already maintains that records the same state (ledger, status file, folded report), or `none`. When set, `target_discovery_command` reads or folds that artifact instead of re-querying the live source, and live calls are reserved for entries the artifact marks as needing action.
 
 Interview rules:
 
 - Ask for missing `success_criteria` before drafting — don't infer it from a vague goal.
 - Ask for missing edge cases when `human_only_blockers`, `evidence_sources`, `fail_condition_rule`, or `write_mode` would materially change behavior.
+- Ask whether a `state_artifact` already exists before accepting a live `target_discovery_command` — re-deriving state a worker already records is the expensive default, not the neutral one.
 
 Before drafting, post a short resolved summary: the filled fields, any defaults taken, open questions if any remain, and a direct check like "Ready to draft?"
 
@@ -68,6 +70,8 @@ Rules:
 - Keep `Goal`, `Motivation`, success rules, fail rules, and blockers concrete, not aspirational.
 - Record assumptions explicitly instead of hiding them.
 - Say what the live target is, how it's rebuilt each round, and how the loop dedupes it — don't hide mutable-state risk.
+- When `state_artifact` is set, name it in `Real target` and `Evidence sources`, and make the per-round read a fold/digest of that artifact — each round's read stays bounded in output size, not a full-history re-parse or a live sweep that streams raw results into a session transcript. Reserve live queries for entries the artifact marks as needing action.
+- A watch loop states its terminal condition in `Exit conditions` — the loop ends when the watched scope reaches it (e.g. the stack merges), not when a human remembers to stop it.
 - If the loop will run unattended, log one row per iteration with `show-me-your-work` instead of inventing a second trail format.
 
 ## Driver shell script contract
@@ -77,6 +81,7 @@ The generated driver must:
 - parse `--target <id>` (repeatable), `--state-file <path>`, `--skip-local-check`, and `--help`;
 - print loop context on start: cwd, branch (if applicable), and the state-file path;
 - rebuild the live target set from `target_discovery_command` every run — never trust a cached list from a prior round;
+- when `state_artifact` is set, implement that rebuild as a bounded fold/digest read of the artifact (latest state per target), not a full-history re-parse and not a live sweep of the underlying source;
 - dedupe the target set by `target_identity_key`;
 - print a repeated-failure summary keyed by `fail_condition_rule`, so a stuck target is visible instead of silently retried forever;
 - when `--target` is passed for inspection, run any dry-run/probe command against a **copy** of mutable state, never the live state file;
diff --git a/product/skills/loop-generator/tests/fires_example.md b/product/skills/loop-generator/tests/fires_example.md
index c0f4f23b4..261d43564 100644
--- a/product/skills/loop-generator/tests/fires_example.md
+++ b/product/skills/loop-generator/tests/fires_example.md
@@ -6,3 +6,10 @@ This matches the trigger phrase "retry failed jobs until they pass" and
 asks for a recurring watch/retry behavior, so the skill runs its
 interview (loop_name, goal, target_discovery_command, fail_condition_rule,
 write_mode, etc.) before drafting the instruction doc + driver script pair.
+
+During the interview the skill also asks whether a `state_artifact` exists
+— e.g. a CI status file or ledger a worker already maintains. If the user
+answers yes, the generated doc names that artifact in `Real target` and
+`Evidence sources`, the per-round read folds it instead of re-querying the
+live source, and the `Exit conditions` section states the terminal
+condition that ends the watch.