merge queue: checking #1360 on main (d89f0c8)

.env.example

@@ -17,6 +17,7 @@ CATSTACK_CAT_MODE_DEFAULT=on
 # Hooks read it on every run; the rule is installed or removed only when
 # ./install.sh runs. 0 (or absent) turns it off.
 CATSTACK_REFLECT_ENFORCEMENT=0
+CATSTACK_HANDBACK_ENFORCEMENT=0
 
 # Everything below is optional. Same lookup order as above; the full table is
 # in engine/hooks/_flags/README.md.

engine/hooks/_flags/README.md

@@ -13,6 +13,7 @@ them go through `flags.py`, so every place works for every flag.
 | `CATSTACK_CAT_MODE_DEFAULT` | `off` (default), `decide`, `on` | When cat-mode applies: only on `/cat-mode`, when the model picks it, or on every prompt. Re-run `./install.sh` after switching to or from `decide`. |
 | `CATSTACK_CAT_MODE_STOP_AFTER_ANSWER` | `off` (default), `on` | Answering the opening question is a stopping point under cat-mode. |
 | `CATSTACK_REFLECT_ENFORCEMENT` | `0` (default), `1` | The reflect/automate-me hooks and rule, described below. |
+| `CATSTACK_HANDBACK_ENFORCEMENT` | `0` (default), `1` | Runs the hand-back check (`handback-needs-attempt`) on its own, without turning on reflect enforcement. Reflect enforcement also turns it on. |
 | `CATSTACK_UNVERIFIED_TAG_BEHAVIOR` | `stale` (default), `off`, `all`, `do_not_emit` | How the unverified-tag ledger reminds you, or refuses tags outright. See `../unverified-tag-ledger/README.md`. |
 | `CATSTACK_HOOK_FRESHNESS` | `local` (default), `fetch`, `off` | Whether hook-freshness checks for stale installed hooks, and whether it may fetch `origin` to do it. |
 | `CATSTACK_HOOK_MODE_<HOOK>` | `off`, `warn`, `stop` | Overrides one hook's mode from `hooks.toml` on this machine, e.g. `CATSTACK_HOOK_MODE_DIU_STOP=warn`. |

engine/hooks/_flags/flags.py

@@ -57,13 +57,15 @@
 ENV_LINE_RE = re.compile(r"^\s*(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(.*?)\s*$")
 
 REFLECT_ENFORCEMENT = "CATSTACK_REFLECT_ENFORCEMENT"
+HANDBACK_ENFORCEMENT = "CATSTACK_HANDBACK_ENFORCEMENT"
 HOOK_DISPATCHER = "CATSTACK_HOOK_DISPATCHER"
 HOOK_MODE_PREFIX = "CATSTACK_HOOK_MODE_"
 
 BEHAVIOR_FLAGS = (
     "CATSTACK_CAT_MODE_DEFAULT",
     "CATSTACK_CAT_MODE_STOP_AFTER_ANSWER",
     REFLECT_ENFORCEMENT,
+    HANDBACK_ENFORCEMENT,
     "CATSTACK_UNVERIFIED_TAG_BEHAVIOR",
     "CATSTACK_HOOK_FRESHNESS",
     HOOK_MODE_PREFIX,
@@ -265,6 +267,37 @@ class on through its own `.env`. None is fine -- the process environment
     return found.on
 
 
+def handback_enforcement(environ: dict, cwd: str | None, home: str | None = None) -> FlagLookup:
+    """Full lookup for the hand-back check's own switch.
+
+    The check judges whether a reply hands the user a step the assistant could
+    have run itself. It does not push anyone into /reflect, so it has its own
+    key and can be on while reflect enforcement stays off.
+    """
+    return resolve_flag(HANDBACK_ENFORCEMENT, environ, cwd, home)
+
+
+def handback_gate(
+    hook_name: str,
+    cwd: str | None,
+    environ: dict | None = None,
+    stderr=None,
+) -> bool:
+    """True when the hand-back switch or reflect enforcement is on.
+
+    Same three-outcome shape as `enforcement_gate`: an unreadable candidate
+    file is named on stderr before the hook goes quiet.
+    """
+    env = os.environ if environ is None else environ
+    own = handback_enforcement(env, cwd)
+    reflect = reflect_enforcement(env, cwd)
+    for found, key in ((own, HANDBACK_ENFORCEMENT), (reflect, REFLECT_ENFORCEMENT)):
+        note = found.unreadable_note(key)
+        if note:
+            (stderr or sys.stderr).write(f"{hook_name}: {note}\n")
+    return own.on or reflect.on
+
+
 def main(argv: list[str] | None = None, environ: dict | None = None, stdout=None, stderr=None) -> int:
     """Print `on`, `off`, or `unchecked` for one key, for callers that are not
     Python. install.sh reads it to pick which always-on rules to install.

engine/hooks/handback-needs-attempt/README.md

@@ -33,7 +33,7 @@ assistant reply, builds a phrase-dictionary job, and sends it to `llm-judge`.
 The dictionary defines the meaning with `match` and `not_match` examples and
 supplies the static `on_hit` follow-up text.
 
-No job is sent when `stop_hook_active` is set, when reflect enforcement is off,
+No job is sent when `stop_hook_active` is set, when both `CATSTACK_HANDBACK_ENFORCEMENT` and reflect enforcement are off,
 or when there is no readable transcript and non-empty assistant reply. A tool
 result does not start a new turn, so a denial or human-only blocker remains in
 the exchange the judge sees. Harness metadata and subagent rows do not count as

engine/hooks/handback-needs-attempt/detect.py

@@ -15,7 +15,7 @@
 sys.path.insert(0, os.path.join(os.path.dirname(HOOKS_DIR), "_flags"))
 sys.path.insert(0, os.path.join(os.path.dirname(HOOKS_DIR), "_sdk"))
 
-from flags import enforcement_gate
+from flags import handback_gate
 from transcripts import resolve_model
 
 CHECKER = "handback-needs-attempt"
@@ -188,7 +188,7 @@ def enqueue_judge(payload: dict, harness: str = "unknown", stderr=None) -> str |
     payload = payload if isinstance(payload, dict) else {}
     if payload.get("stop_hook_active"):
         return None
-    if not enforcement_gate(CHECKER, payload.get("cwd"), stderr=stderr or sys.stderr):
+    if not handback_gate(CHECKER, payload.get("cwd"), stderr=stderr or sys.stderr):
         return None
     path = resolve_transcript(payload)
     reply = last_assistant_text(payload, path)

engine/hooks/handback-needs-attempt/tests/test_hooks.py

@@ -192,6 +192,28 @@ def test_stop_hook_active_stays_silent(self):
         self.assertIsNone(detect.enqueue_judge({**self.payload(HANDOFF_RUN, path), "stop_hook_active": True}, "claude"))
         self.assertEqual(self.jobs(), [])
 
+    def test_own_switch_runs_the_judge_without_reflect_enforcement(self):
+        self.use_runners(ANSWERS_HIT)
+        path = self.transcript(human_line("set up Slack"))
+        with patch.dict(os.environ, {
+            "CATSTACK_REFLECT_ENFORCEMENT": "0",
+            "CATSTACK_HANDBACK_ENFORCEMENT": "1",
+        }):
+            job_id = detect.enqueue_judge(self.payload(HANDOFF_RUN, path), "claude")
+        self.assertIsNotNone(job_id)
+        messages = self.wait_for_messages(path)
+        self.assertTrue(any("handback-needs-attempt" in message for message in messages))
+
+    def test_silent_when_both_switches_are_off(self):
+        path = self.transcript(human_line("set up Slack"))
+        with patch.dict(os.environ, {
+            "CATSTACK_REFLECT_ENFORCEMENT": "0",
+            "CATSTACK_HANDBACK_ENFORCEMENT": "0",
+        }):
+            job_id = detect.enqueue_judge(self.payload(HANDOFF_RUN, path), "claude")
+        self.assertIsNone(job_id)
+        self.assertEqual(self.jobs(), [])
+
     def test_stop_wrapper_never_blocks_the_reply(self):
         self.use_runners(ANSWERS_CLEAN)
         path = self.transcript(human_line("set up Slack"))