merge queue: checking #1360 on main (2f93d6c), stacked on 4 pull requests ahead

.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.

corpus/CLAUDE.learned.md

@@ -39,6 +39,7 @@ Engine-only install drops this file; reflect Accepted global rules land here.
 
 - After an audit that names existing scorers, rankers, or weight formulas, the next plan picks among those named gaps. Bind "real classifier" / "these weights are arbitrary" onto a named scorer before proposing a new model or library. If I say I'm not comfortable with ML, teach the existing named scorers first and offer a no-library path before sklearn. The audit's named gaps are the backlog: a plan that jumps to a new classifier throws away findings already paid for, and the fix the audit points at (for example, scoring a signal by how many events it appears in rather than by its exact string) is usually a small change to a scorer that exists. This is Zinkevich's first rule of machine learning: do not be afraid to launch without machine learning — improve the heuristic you have before replacing it.
 - When a permission classifier or sandbox denies a tool call, treat the first denial as categorical for that target — don't retry the same blocked action through a different tool (Bash → Edit → a different Bash invocation). Switch strategy immediately, usually to a script handoff (next rule). The policy judges the target and the action, not the tool, so every route to the same write meets the same check — Saltzer and Schroeder's complete mediation — and hunting for an unguarded route is working around a security control.
+- Before handing me a step to run, run its first command yourself and read the result. Hand it back only when the result names a blocker that only I can clear: a password, a browser login, hardware, or an operating-system approval dialog. A missing tool, a wrong path, or a script you have not run is not a blocker: install the tool, fix the path, or run the script, then report. When you do hand a step back, quote the blocker in the same message. A handoff written before the first command was tried tends to fail on its own first line, and the person pays for it twice.
 - When handing off a blocked action for me to run myself, always write it to a small script file and hand back exactly one `! bash <path>` / `! python3 <path>` line — never paste inline multi-line, multi-flag, or `&&`/`&`-backgrounded shell text into chat for me to copy. This applies on the very first handoff and every one after it, not after I complain about copy-paste pain. Pasted shell breaks in transit — wrapped lines, lost quoting, an `&` that detaches a job — while one `!` line runs exactly the file that was written.
 - Never pass a quoted command string through `sudo -i`, `su -`, or `su -l`. Those start a login shell that reads the command a second time, after the first shell already used up the quotes, so the string splits on spaces and its first word runs alone (`sudo -u demo -H -i bash -lc 'set -u; echo one'` fails with `set: -c: invalid option`; the same line without `-i` prints `one`). Copy the script to the host and run it by path, or drop `-i`.
 - The shell behind the Bash tool may be zsh, which does not split an unquoted variable into words: `for x in $LIST` runs once with the whole list as one item, exits 0, and reads as "all done". Run any loop over a list under bash (`bash <<'EOF' ... EOF`), count lookups that failed as unchecked rather than done, and spot-check one item before acting on the loop's summary. This is zsh's `SH_WORD_SPLIT` option being off by default (zsh manual, Options, https://zsh.sourceforge.io/Doc/Release/Options.html#index-SH_005fWORD_005fSPLIT).

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/_runner/tests/test_wrap_installed.py

@@ -24,6 +24,9 @@ def setUp(self):
         self.old_home = os.environ.get("HOME")
         os.environ["HOME"] = str(self.home)
         self.addCleanup(self._restore_home)
+        dispatcher_patch = mock.patch.dict(os.environ, {"CATSTACK_HOOK_DISPATCHER": "0"})
+        dispatcher_patch.start()
+        self.addCleanup(dispatcher_patch.stop)
         self.claude_path = self.home / ".claude" / "settings.json"
         self.cursor_path = self.home / ".cursor" / "hooks.json"
         self.codex_path = self.home / ".codex" / "hooks.json"

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"))

product/skills/ci-job-timing/SKILL.md

@@ -0,0 +1,43 @@
+---
+name: ci-job-timing
+description: >-
+  Repeatable per-step timing report for any GitLab CI job, from Datadog. Use
+  when asked how long a CI job's steps take, where a slow job spends its time,
+  or to compare the same job across dev/prod or across different jobs
+  (for example "Build LAGVM (dev)", "xmm-image-packager (dev)").
+---
+
+# ci-job-timing
+
+One recipe, any job. The job name and the number of days are the only inputs.
+
+## Why it reads log text
+
+Datadog CI Visibility has job-level events (`@ci.job.name`, `@duration` in
+nanoseconds) but no step-level events. Step times come from `[CI_STEP]` log
+lines written by the CI step wrapper. Lines from side-by-side tracks carry a
+prefix (for example `[gvm]`, `[qssi]`) and Datadog does not split them into
+fields, so the script reads the message text ("took N seconds") instead.
+
+## Steps
+
+1. Build the query and its link (`<skill>` is this skill's folder):
+   `python3 <skill>/scripts/ci_job_timing.py queries --job "<exact job name>" --days 3`
+2. Run the printed steps query with the Datadog `ddsql_run_query` tool (from `now-<days>d`).
+3. Save the TSV it returns to a file and render the table:
+   `python3 <skill>/scripts/ci_job_timing.py report --job "<job>" --days 3 --tsv <file>`
+4. For side-by-side steps, pick one `job_id` from a raw `[CI_STEP]` line and run
+   `ci_job_timing.py timeline --job "<job>" --job-id <id>`. Read the start and end
+   times there; never add step averages into a job time.
+5. Optional job-level cross-check: `aggregate_datadog_ci_pipeline_events` with
+   `ci_level=job`, query `@ci.job.name:"<job>"`, metric `@duration`.
+
+## Read the table correctly
+
+- `Runs` counts successful step lines only. A run that stopped early does not
+  count in its later steps, so counts differ between steps.
+- Averages with few runs (n under 10) are weak. Say the n.
+- Steps can overlap, so the column is not a share of the job.
+- Zero rows means a wrong job name or an empty window: the script errors, it
+  never prints an empty table.
+- A job whose steps do not log `[CI_STEP]` lines returns nothing; say that.

product/skills/ci-job-timing/scripts/ci_job_timing.py

@@ -0,0 +1,139 @@
+#!/usr/bin/env python3
+"""Build the Datadog queries and the report for one CI job's step timing.
+
+The Datadog connector is only reachable from the agent, so this script does the
+two parts that must be identical on every run: write the query text (`queries`,
+`timeline`) and turn the query result into a table (`report`).
+"""
+from __future__ import annotations
+
+import argparse
+import csv
+import io
+import re
+import sys
+from urllib.parse import quote
+
+DEFAULT_HOST = "rivianvw.datadoghq.com"
+# GitLab job names look like "Build LAGVM (dev)": letters, digits, space, ( ) . _ : / -
+_JOB_RE = re.compile(r"^[A-Za-z0-9 ()._:/\-]+$")
+
+
+def check_job(job: str) -> str:
+    if not job or not _JOB_RE.match(job):
+        raise ValueError(f"job name has characters the query cannot carry safely: {job!r}")
+    return job
+
+
+def check_days(days: int) -> int:
+    if not 1 <= days <= 30:
+        raise ValueError(f"days must be 1-30, got {days}")
+    return days
+
+
+def _seconds_expr() -> str:
+    return "CAST(SPLIT_PART(SPLIT_PART(message, 'took ', 2), ' seconds', 1) AS DECIMAL)"
+
+
+def steps_sql(job: str, days: int) -> str:
+    """One row per step: runs, average, min, max seconds. Reads message text, so
+    lines prefixed [gvm]/[qssi] (which Datadog does not split into fields) count too."""
+    check_job(job)
+    check_days(days)
+    flt = f'"[CI_STEP]" STEP_SUCCESS "[job={job}]"'
+    sec = _seconds_expr()
+    return (
+        "SELECT SPLIT_PART(SPLIT_PART(message, 'Step: <', 2), '>', 1) AS step, "
+        f"COUNT(*) AS runs, AVG({sec}) AS avg_s, MIN({sec}) AS min_s, MAX({sec}) AS max_s "
+        f"FROM dd.logs(columns => ARRAY['message'], filter => '{flt}', "
+        f"from_timestamp => NOW() - INTERVAL '{days} day', to_timestamp => NOW()) "
+        "AS (message VARCHAR) GROUP BY SPLIT_PART(SPLIT_PART(message, 'Step: <', 2), '>', 1) "
+        "ORDER BY avg_s DESC LIMIT 60"
+    )
+
+
+def timeline_sql(job: str, job_id: int, days: int) -> str:
+    """Every step line of one run, oldest first, so side-by-side steps are visible."""
+    check_job(job)
+    check_days(days)
+    flt = f'"[CI_STEP]" "[job_id={int(job_id)}]" "[job={job}]"'
+    return (
+        "SELECT timestamp, message "
+        f"FROM dd.logs(columns => ARRAY['timestamp', 'message'], filter => '{flt}', "
+        f"from_timestamp => NOW() - INTERVAL '{days} day', to_timestamp => NOW()) "
+        "AS (timestamp TIMESTAMP, message VARCHAR) ORDER BY timestamp LIMIT 200"
+    )
+
+
+def link(sql: str, host: str = DEFAULT_HOST) -> str:
+    return f"https://{host}/ddsql/editor?q={quote(sql, safe='')}"
+
+
+def parse_tsv(text: str) -> list[dict]:
+    """Parse the TSV block ddsql_run_query returns. Missing header or no rows -> error."""
+    body = text
+    m = re.search(r"<TSV_DATA>(.*?)</TSV_DATA>", text, re.S)
+    if m:
+        body = m.group(1)
+    body = body.strip("\n")
+    if not body.strip():
+        raise ValueError("no rows: the query returned nothing (wrong job name, or no runs in the window)")
+    rows = list(csv.DictReader(io.StringIO(body), delimiter="\t"))
+    need = {"step", "runs", "avg_s", "min_s", "max_s"}
+    if not rows or not need <= set(rows[0]):
+        raise ValueError(f"unexpected columns, need {sorted(need)}; got {list(rows[0]) if rows else 'none'}")
+    return rows
+
+
+def report(job: str, days: int, rows: list[dict], url: str) -> str:
+    out = [
+        f"## {job}: step times, last {days} day(s)",
+        "",
+        "| Step | Runs | Avg (s) | Avg (min) | Fastest (s) | Slowest (s) |",
+        "|---|---|---|---|---|---|",
+    ]
+    for r in rows:
+        avg = float(r["avg_s"])
+        out.append(
+            f"| {r['step']} | {int(float(r['runs']))} | {avg:.0f} | {avg / 60:.1f} "
+            f"| {float(r['min_s']):.0f} | {float(r['max_s']):.0f} |"
+        )
+    out += [
+        "",
+        "Steps can run side by side, so do not add these up to a job time.",
+        f"Query: {url}",
+    ]
+    return "\n".join(out)
+
+
+def main(argv: list[str] | None = None) -> int:
+    ap = argparse.ArgumentParser(description=__doc__)
+    sub = ap.add_subparsers(dest="cmd", required=True)
+    for name in ("queries", "report", "timeline"):
+        p = sub.add_parser(name)
+        p.add_argument("--job", required=True)
+        p.add_argument("--days", type=int, default=3)
+        p.add_argument("--host", default=DEFAULT_HOST)
+        if name == "timeline":
+            p.add_argument("--job-id", type=int, required=True)
+        if name == "report":
+            p.add_argument("--tsv", required=True, help="file holding the steps query result")
+    a = ap.parse_args(argv)
+    try:
+        if a.cmd == "queries":
+            sql = steps_sql(a.job, a.days)
+            print(f"-- steps query\n{sql}\n-- link\n{link(sql, a.host)}")
+        elif a.cmd == "timeline":
+            sql = timeline_sql(a.job, a.job_id, a.days)
+            print(f"-- timeline query\n{sql}\n-- link\n{link(sql, a.host)}")
+        else:
+            rows = parse_tsv(open(a.tsv, encoding="utf-8").read())
+            print(report(a.job, a.days, rows, link(steps_sql(a.job, a.days), a.host)))
+    except (ValueError, OSError) as e:
+        print(f"error: {e}", file=sys.stderr)
+        return 2
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

product/skills/ci-job-timing/tests/test_ci_job_timing.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
+
+import ci_job_timing as c  # noqa: E402
+
+TSV = """<METADATA><total_rows>2</total_rows></METADATA>
+<TSV_DATA>
+step\truns\tavg_s\tmin_s\tmax_s
+Build LAGVM System\t4\t1563.0\t1500\t1600
+Build LAGVM Kernel\t16\t295.0\t250\t446
+</TSV_DATA>"""
+
+
+class StepsSql(unittest.TestCase):
+    def test_job_and_days_are_in_the_query(self):
+        sql = c.steps_sql("Build LAGVM (dev)", 3)
+        self.assertIn('"[job=Build LAGVM (dev)]"', sql)
+        self.assertIn("INTERVAL '3 day'", sql)
+
+    def test_same_template_works_for_another_job(self):
+        sql = c.steps_sql("xmm-image-packager (dev)", 7)
+        self.assertIn("[job=xmm-image-packager (dev)]", sql)
+
+    def test_quote_in_job_name_is_refused(self):
+        with self.assertRaises(ValueError):
+            c.steps_sql("x' OR 1=1 --", 3)
+
+    def test_days_out_of_range_is_refused(self):
+        with self.assertRaises(ValueError):
+            c.steps_sql("Build LAGVM (dev)", 0)
+
+
+class Report(unittest.TestCase):
+    def test_parses_and_renders(self):
+        rows = c.parse_tsv(TSV)
+        out = c.report("Build LAGVM (dev)", 3, rows, "https://example/x")
+        self.assertIn("| Build LAGVM System | 4 | 1563 | 26.1 | 1500 | 1600 |", out)
+
+    def test_empty_result_is_an_error_not_an_empty_table(self):
+        with self.assertRaises(ValueError):
+            c.parse_tsv("<TSV_DATA>\n</TSV_DATA>")
+
+    def test_wrong_columns_is_an_error(self):
+        with self.assertRaises(ValueError):
+            c.parse_tsv("<TSV_DATA>\nfoo\tbar\n1\t2\n</TSV_DATA>")
+
+    def test_timeline_uses_job_id(self):
+        self.assertIn("[job_id=253261575]", c.timeline_sql("Build LAGVM (dev)", 253261575, 3))
+
+
+if __name__ == "__main__":
+    unittest.main()