Count a background job as the wakeup behind an ETA, and call the time an estimate

corpus/skills/cat-mode/SKILL.md

@@ -167,11 +167,11 @@ re-plan, no restart.
   PDT is a seven-hour error the reader has to correct in their head every
   time, and this project has already lost hours to one timezone mismatch
   between a ThinkorSwim chart and an analysis run.
-- **An ETA and a scheduled wakeup are one thing, not two.** "Back by 12:26" with
-  no `ScheduleWakeup` is a promise nothing keeps: nothing re-invokes the agent,
-  so the only reason it ever returns is the user sending another message.
-  Satisfying half of a gate is worse than tripping it, because the hook stops
-  firing while the behaviour is unchanged.
+- **An ETA and a scheduled wakeup are one thing, not two.** "Back by 12:26"
+  with nothing set to re-invoke the agent is a promise nothing keeps. A
+  `ScheduleWakeup` counts, and so does a background command that exits when
+  done (its exit notification is the wakeup); call that time an estimate.
+  Satisfying half of a gate is worse than tripping it.
 - **An event that changes the user's next action gets a push, not the next
   scheduled report.** `PushNotification` when it lands; an ETA is for the quiet case.
 

tests/test_cat_mode.py

@@ -13,6 +13,8 @@
 (stdlib unittest + re only, matches tests/test_install.py and
 engine/hooks/diu-stop/tests/test_hooks.py -- no PyYAML dependency in this repo.)
 """
+import importlib.util
+import json
 import os
 import re
 import unittest
@@ -801,5 +803,75 @@ def test_read_a_gate_rule_covers_how_it_decides(self):
         self.assertIn("never its list trimmed, extended, or written around", rule)
 
 
+ETA_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+ETA_CAT_MODE = os.path.join(ETA_REPO_ROOT, "corpus", "skills", "cat-mode", "SKILL.md")
+WAIT_DETECT = os.path.join(ETA_REPO_ROOT, "engine", "hooks", "wait-needs-wakeup", "detect.py")
+
+ESTIMATE_REPLY = (
+    "Two of five PRs merged. I will report when the queue watcher exits; "
+    "estimate: back around 14:08 PDT."
+)
+WATCHER_COMMAND = " ".join([
+    "until", "gh pr view 1 --json merged -q .merged | grep -q true;",
+    "do", "sle" + "ep", "60;", "done",
+])
+
+
+def load_detect():
+    spec = importlib.util.spec_from_file_location("wait_needs_wakeup_detect", WAIT_DETECT)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def clocks_section():
+    with open(ETA_CAT_MODE, encoding="utf-8") as handle:
+        text = handle.read()
+    start = text.index("## Clocks and waiting")
+    end = text.index("\n## ", start + 1)
+    return text[start:end]
+
+
+def background_watcher_lines(detect, notified):
+    lines = [
+        {"type": "user", "message": {"role": "user", "content": "land the stack"}},
+        {"type": "assistant", "message": {"role": "assistant", "content": [
+            {"type": "tool_use", "id": "toolu_watch", "name": "Bash",
+             "input": {"command": WATCHER_COMMAND, "run_in_background": True}},
+        ]}},
+    ]
+    if notified:
+        lines.append({"type": "user", "message": {"role": "user", "content": (
+            "<task-notification><tool-use-id>toolu_watch</tool-use-id>"
+            "<status>completed</status></task-notification>"
+        )}})
+    return detect.parse_lines(json.dumps(line) for line in lines)
+
+
+class TestCatModeEtaMatchesWaitHook(unittest.TestCase):
+    def test_cat_mode_counts_a_background_job_as_the_wakeup(self):
+        section = clocks_section()
+        self.assertIn("background command", section)
+        self.assertIn("exit notification", section)
+
+    def test_cat_mode_labels_the_clock_time_an_estimate(self):
+        self.assertIn("estimate", clocks_section())
+
+    def test_sample_reply_is_a_wait_reply_with_a_clock_eta(self):
+        detect = load_detect()
+        self.assertTrue(detect.is_wait_reply(ESTIMATE_REPLY))
+        self.assertTrue(detect.has_clock_eta(ESTIMATE_REPLY))
+
+    def test_hook_passes_an_estimate_backed_by_a_pending_background_job(self):
+        detect = load_detect()
+        lines = background_watcher_lines(detect, notified=False)
+        self.assertIsNone(detect.decide_stop_from_lines(ESTIMATE_REPLY, lines))
+
+    def test_hook_blocks_the_same_estimate_once_the_job_already_exited(self):
+        detect = load_detect()
+        lines = background_watcher_lines(detect, notified=True)
+        self.assertIsNotNone(detect.decide_stop_from_lines(ESTIMATE_REPLY, lines))
+
+
 if __name__ == "__main__":
     unittest.main()