NK-WP-0058-T08 worker protocol and excerpt tool; T06 window watch (row O5), first use of the protocol

T08: .claude/rules/worker-protocol.md (imported by CLAUDE.md, pointed to from AGENTS.md) and
tools/contract-tests/excerpt.py: name the rows, hand every worker one excerpt, one owner per step, check
with the excerpt's tests. A test requires every table row to have details and a contract test (found O4).
T06: bounded-executor/window_watch.py flags or closes the write role when no run Job is active or recent;
stub tests; in the image. Built from the excerpt for O5, E1, C1, E8, E9, R2.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 943911@bnt-lap001
Assistant-Session: fc322940-1c8e-4bed-ab39-86cec99f23a0

bounded-executor/tests/test_window_watch.py

@@ -0,0 +1,110 @@
+"""window_watch.py on stubs (NK-WP-0058-T06, contract row O5): rule, both modes, and errors."""
+import datetime as dt
+import importlib.util
+import pathlib
+import sys
+import unittest
+
+HERE = pathlib.Path(__file__).resolve().parent
+sys.path.insert(0, str(HERE.parent))
+spec = importlib.util.spec_from_file_location('window_watch', HERE.parent / 'window_watch.py')
+W = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(W)
+OPS = W.OPS
+NOW = dt.datetime(2026, 11, 6, 3, 0, tzinfo=dt.timezone.utc)
+
+
+def job(minutes_ago, active=0):
+    created = (NOW - dt.timedelta(minutes=minutes_ago)).isoformat().replace('+00:00', 'Z')
+    return {'metadata': {'creationTimestamp': created}, 'status': {'active': active} if active else {}}
+
+
+class FakeBao:
+    def __init__(self, present=True, delete_works=True):
+        self.present, self.delete_works, self.calls = present, delete_works, []
+
+    def _call(self, method, path, body=None, auth=True):
+        self.calls.append((method, path))
+        assert path == W.ROLE_PATH, path  # nothing but the write role is ever touched
+        if method == 'GET':
+            if not self.present:
+                raise OPS.OperationError('openbao GET 404')
+            return {'data': {}}
+        if method == 'DELETE':
+            self.present = not self.delete_works
+            return {}
+        raise AssertionError(method)
+
+
+class FakeKube:
+    def __init__(self, jobs):
+        self.jobs, self.paths = jobs, []
+
+    def get(self, path):
+        self.paths.append(path)
+        return {'items': self.jobs}
+
+
+class Rule(unittest.TestCase):
+    def test_decide(self):
+        cases = [
+            (False, [], 'ok'),                              # no role: nothing open
+            (True, [], 'stale'),                            # role, never a run: left open
+            (True, [job(3)], 'ok'),                         # role, run just created: a window in use
+            (True, [job(30, active=1)], 'ok'),              # role, run still active
+            (True, [job(30)], 'stale'),                     # role, last run finished 30 min ago: left open
+            (True, [job(30), job(2)], 'ok'),                # the newest run counts
+        ]
+        for present, jobs, verdict in cases:
+            with self.subTest(present=present, jobs=len(jobs)):
+                self.assertEqual(W.decide(present, jobs, NOW)[0], verdict)
+
+    def test_run_jobs_selector_excludes_other_components(self):
+        kube = FakeKube([])
+        W.run_jobs(kube)
+        self.assertIn('labelSelector=app.kubernetes.io%2Fname%3Dbounded-executor%2C%21app.kubernetes.io%2Fcomponent',
+                      kube.paths[0])
+        self.assertTrue(kube.paths[0].startswith('/apis/batch/v1/namespaces/sso/jobs?'))
+
+
+class Modes(unittest.TestCase):
+    def test_ok_touches_nothing(self):
+        bao = FakeBao(present=False)
+        result, code = W.watch(bao, FakeKube([]), 'close', NOW)
+        self.assertEqual((result['status'], code), ('ok', 0))
+        self.assertEqual([m for m, _ in bao.calls], ['GET'])
+
+    def test_flag_reads_only_and_fails_the_job(self):
+        bao = FakeBao()
+        result, code = W.watch(bao, FakeKube([job(30)]), 'flag', NOW)
+        self.assertEqual((result['status'], code), ('flagged_stale', 3))
+        self.assertEqual([m for m, _ in bao.calls], ['GET'])
+        self.assertEqual(result['newest_age_s'], 1800)
+
+    def test_close_deletes_and_proves_absent(self):
+        bao = FakeBao()
+        result, code = W.watch(bao, FakeKube([]), 'close', NOW)
+        self.assertEqual((result['status'], code), ('closed_stale', 0))
+        self.assertEqual([m for m, _ in bao.calls], ['GET', 'DELETE', 'GET'])
+        self.assertFalse(bao.present)
+
+    def test_close_that_does_not_take_is_an_error(self):
+        result, code = W.watch(FakeBao(delete_works=False), FakeKube([]), 'close', NOW)
+        self.assertEqual((result['status'], result['reason'], code), ('error', 'write_role_still_present', 5))
+
+    def test_a_window_in_use_is_never_closed(self):
+        bao = FakeBao()
+        result, code = W.watch(bao, FakeKube([job(1, active=1)]), 'close', NOW)
+        self.assertEqual((result['status'], code), ('ok', 0))
+        self.assertTrue(bao.present)
+
+    def test_other_openbao_errors_are_not_read_as_absent(self):
+        class Denied(FakeBao):
+            def _call(self, method, path, body=None, auth=True):
+                raise OPS.OperationError('openbao GET 403')
+        with self.assertRaises(OPS.OperationError):
+            W.watch(Denied(), FakeKube([]), 'close', NOW)
+
+
+if __name__ == '__main__':
+    unittest.main()