hooks: attach X-Cogos-Grant to kernel write requests (v0.16.29 enforcement)

plugins/cogos-harness/hooks/seat-identity-heal.py

@@ -39,6 +39,39 @@
 KERNEL_URL = os.environ.get("COGOS_KERNEL_URL") or \
     f"http://127.0.0.1:{os.environ.get('COGOS_KERNEL_PORT', '6931')}"
 TIMEOUT = 1.0  # seconds; localhost-only call, kept short per the hook budget
+GRANT_TIMEOUT = 1.0
+
+# v0.16.29: kernel writes require an X-Cogos-Grant header. Resolved once per
+# process and cached: vault file first, loopback grants/current GET as
+# fallback. Any acquisition failure means "proceed without the header" --
+# fail-open, matching this hook's own silent-no-op contract. A broken vault
+# must never break the heal; the 401 that follows routes to the existing
+# fallback additionalContext exactly as a kernel-down response would.
+_GRANT_CACHE: dict = {"tried": False, "token": None}
+
+
+def _get_grant() -> str | None:
+    if _GRANT_CACHE["tried"]:
+        return _GRANT_CACHE["token"]
+    _GRANT_CACHE["tried"] = True
+    token = None
+    try:
+        raw = (Path.home() / ".cog" / "vault" / "node-root-grant").read_text(encoding="utf-8")
+        token = raw.strip() or None
+    except Exception:
+        token = None
+    if not token:
+        try:
+            with urllib.request.urlopen(
+                f"{KERNEL_URL}/v1/identity/grants/current?surface=node-root",
+                timeout=GRANT_TIMEOUT,
+            ) as r:
+                if r.status == 200:
+                    token = (json.loads(r.read()).get("token")) or None
+        except Exception:
+            token = None
+    _GRANT_CACHE["token"] = token
+    return token
 
 
 def _load_identity() -> dict | None:
@@ -69,10 +102,14 @@ def _register(identity: dict) -> dict | None:
     }
     if identity.get("hostname"):
         body["hostname"] = identity["hostname"]
+    headers = {"Content-Type": "application/json"}
+    grant = _get_grant()
+    if grant:
+        headers["X-Cogos-Grant"] = grant
     req = urllib.request.Request(
         f"{KERNEL_URL}/v1/sessions/register",
         data=json.dumps(body).encode(),
-        headers={"Content-Type": "application/json"},
+        headers=headers,
         method="POST",
     )
     try:

plugins/cogos-harness/hooks/user-scope-session-end.py

@@ -39,6 +39,40 @@
     f"http://127.0.0.1:{os.environ.get('COGOS_KERNEL_PORT', '6931')}"
 PROBE_TIMEOUT = 1.0
 END_TIMEOUT = 1.5
+GRANT_TIMEOUT = 1.0
+
+# v0.16.29: kernel writes require an X-Cogos-Grant header. Resolved once per
+# process and cached: vault file first, loopback grants/current GET as
+# fallback. Any acquisition failure means "proceed without the header" --
+# fail-open, same contract as the rest of this hook. A broken vault must
+# never break session-end; the 401 that follows lands in the existing
+# fire-and-forget error handling exactly as it did before this header
+# existed.
+_GRANT_CACHE: dict = {"tried": False, "token": None}
+
+
+def _get_grant() -> str | None:
+    if _GRANT_CACHE["tried"]:
+        return _GRANT_CACHE["token"]
+    _GRANT_CACHE["tried"] = True
+    token = None
+    try:
+        raw = (Path.home() / ".cog" / "vault" / "node-root-grant").read_text(encoding="utf-8")
+        token = raw.strip() or None
+    except Exception:
+        token = None
+    if not token:
+        try:
+            with urllib.request.urlopen(
+                f"{KERNEL_URL}/v1/identity/grants/current?surface=node-root",
+                timeout=GRANT_TIMEOUT,
+            ) as r:
+                if r.status == 200:
+                    token = (json.loads(r.read()).get("token")) or None
+        except Exception:
+            token = None
+    _GRANT_CACHE["token"] = token
+    return token
 
 
 def _find_cog_workspace() -> Path | None:
@@ -120,13 +154,17 @@ def _end_direct(session_id: str) -> None:
     cog_end_session. Fire-and-forget, fail-open on any error (unknown
     session -> 404, already-ended -> 409, kernel down -> connection error
     -- all silently ignored, matching the rest of this hook's contract)."""
+    headers = {"Content-Type": "application/json"}
+    grant = _get_grant()
+    if grant:
+        headers["X-Cogos-Grant"] = grant
     req = urllib.request.Request(
         f"{KERNEL_URL}/v1/sessions/{urllib.parse.quote(session_id, safe='')}/end",
         # "session_end_hook" is the fleet-wide end_reason vocabulary
         # (established by the settings.local.json session-awareness hook);
         # consumers key on it, so the plugin fallback must not fork it.
         data=json.dumps({"reason": "session_end_hook"}).encode(),
-        headers={"Content-Type": "application/json"},
+        headers=headers,
         method="POST",
     )
     try:

plugins/cogos-harness/hooks/user-scope-session-start.py

@@ -67,6 +67,40 @@
     f"http://127.0.0.1:{os.environ.get('COGOS_KERNEL_PORT', '6931')}"
 PROBE_TIMEOUT = 1.0  # short, bounded probe -- never block the hook budget
 REGISTER_TIMEOUT = 1.5
+GRANT_TIMEOUT = 1.0
+
+# v0.16.29: kernel writes require an X-Cogos-Grant header. Resolved once per
+# process and cached: vault file first, loopback grants/current GET as
+# fallback. Any acquisition failure means "proceed without the header" --
+# fail-open, same contract as the rest of this hook. A broken vault must
+# never break registration; the 401 that follows lands in the existing
+# fire-and-forget error handling exactly as it did before this header
+# existed.
+_GRANT_CACHE: dict = {"tried": False, "token": None}
+
+
+def _get_grant() -> str | None:
+    if _GRANT_CACHE["tried"]:
+        return _GRANT_CACHE["token"]
+    _GRANT_CACHE["tried"] = True
+    token = None
+    try:
+        raw = (Path.home() / ".cog" / "vault" / "node-root-grant").read_text(encoding="utf-8")
+        token = raw.strip() or None
+    except Exception:
+        token = None
+    if not token:
+        try:
+            with urllib.request.urlopen(
+                f"{KERNEL_URL}/v1/identity/grants/current?surface=node-root",
+                timeout=GRANT_TIMEOUT,
+            ) as r:
+                if r.status == 200:
+                    token = (json.loads(r.read()).get("token")) or None
+        except Exception:
+            token = None
+    _GRANT_CACHE["token"] = token
+    return token
 
 
 def _find_cog_workspace() -> Path | None:
@@ -196,10 +230,14 @@ def _register_direct(session_id: str, workspace: str, source: str = "") -> None:
         "hostname": socket.gethostname(),
         "extras": {"source": source or "startup", "via": "cogos-harness"},
     }
+    headers = {"Content-Type": "application/json"}
+    grant = _get_grant()
+    if grant:
+        headers["X-Cogos-Grant"] = grant
     req = urllib.request.Request(
         f"{KERNEL_URL}/v1/sessions/register",
         data=json.dumps(body).encode(),
-        headers={"Content-Type": "application/json"},
+        headers=headers,
         method="POST",
     )
     try: