Merge feature/workbench-w1-operator-tasks into unify/skills-20261112

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

orchestrator/kvb_orchestrator/asking.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import time
 
 from .choosing import EVERYONE, answer_text
+from .diffs import store_diff, valid_diff
 from .errors import Conflict, Missing, ToolError
 from .fields import (
     COMMAND_LIMIT,
@@ -60,12 +61,16 @@ def option_of(option: object) -> dict:
 class Asking:
     def ask(self, body: bytes) -> dict:
         payload = decode(body)
+        diff = valid_diff(payload.pop("diff")) if "diff" in payload else None
         expect(payload, ("tool", *ASK_TEXTS, "timeout"))
         values = {name: bounded(payload, name, limit) for name, limit in ASK_TEXTS.items()}
         lifetime = seconds(payload, "timeout", ASK_LIFETIME)
         created = time.time()
-        fields = {**values, "tool": field(payload, "tool").strip(), "status": "pending", "decided": None, "proof": None}
-        record = self.asks.add({**fields, "created_ts": created, "expires": created + lifetime})
+        fields = {**values, "tool": field(payload, "tool").strip(), "status": "pending", "decided": None, "proof": None, "diff_files": None if diff is None else len(diff)}
+        ask_id = self.asks.fresh()
+        if diff is not None:
+            store_diff(self.reviews.pages, "ask", ask_id, diff)
+        record = self.asks.insert(ask_id, {**fields, "created_ts": created, "expires": created + lifetime})
         self.automate("ask", record)
         return {"id": record["id"], "status": record["status"]}
 

orchestrator/kvb_orchestrator/diffs.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import json
+import re
+from pathlib import Path
+
+from .errors import Missing, ToolError
+from .records import read_json, write_atomic
+from .signed import limits
+
+DIFF_LIMIT = limits.DIFF_LIMIT
+DIFF_BODY_LIMIT = limits.DIFF_BODY_LIMIT
+FILE_CAP = limits.FILE_CAP
+PATH_LIMIT = limits.PATH_LIMIT
+STATUSES = ("A", "M", "D", "R", "B")
+DIFF_KEYS = ("path", "status", "mode", "additions", "deletions", "patch", "note")
+NOTES = ("", limits.WITHHELD)
+MODE = re.compile(r"[0-7]{6}(?: => [0-7]{6})?")
+DIFF_SUFFIX = {"review": ".review-diff.json", "ask": ".ask-diff.json"}
+
+
+def counted(value: object) -> bool:
+    return isinstance(value, int) and not isinstance(value, bool) and value >= 0
+
+
+def diff_entry(entry: object) -> dict:
+    if not isinstance(entry, dict) or set(entry) != set(DIFF_KEYS):
+        raise ToolError(f"every diff entry must be an object with exactly {', '.join(DIFF_KEYS)}")
+    path, patch, mode = entry["path"], entry["patch"], entry["mode"]
+    if not isinstance(path, str) or not 0 < len(path) <= PATH_LIMIT or entry["status"] not in STATUSES:
+        raise ToolError(f"a diff path must be 1 to {PATH_LIMIT} characters and its status one of {', '.join(STATUSES)}")
+    if not isinstance(mode, str) or not MODE.fullmatch(mode):
+        raise ToolError("mode must be a git file mode, or old => new for a changed one")
+    if not counted(entry["additions"]) or not counted(entry["deletions"]) or not isinstance(patch, str) or (entry["status"] == "B" and patch):
+        raise ToolError("additions and deletions must be counts, patch text, empty for a binary file")
+    if entry["note"] not in NOTES or (entry["note"] and patch):
+        raise ToolError(f"note must be empty or {limits.WITHHELD!r}, the latter with an empty patch")
+    return {key: entry[key] for key in DIFF_KEYS}
+
+
+def valid_diff(value: object) -> list[dict]:
+    """Returns:
+        The review diff of requirement 003, checked: at most FILE_CAP
+        entries, and at most DIFF_LIMIT bytes of JSON-encoded patch text in
+        all, the bound the requesting script keeps.
+    """
+    if not isinstance(value, list) or len(value) > FILE_CAP:
+        raise ToolError(f"diff must be a list of at most {FILE_CAP} files")
+    entries = [diff_entry(entry) for entry in value]
+    if sum(len(json.dumps(entry["patch"], ensure_ascii=False).encode("utf-8")) for entry in entries) > DIFF_LIMIT:
+        raise ToolError(f"the patches of a diff must stay within {DIFF_LIMIT} bytes")
+    return entries
+
+
+def diff_bytes(diff: list[dict]) -> bytes:
+    return json.dumps({"diff": diff}, ensure_ascii=False).encode("utf-8")
+
+
+def store_diff(folder: Path, kind_name: str, item_id: str, diff: list[dict]) -> None:
+    write_atomic(folder / f"{item_id}{DIFF_SUFFIX[kind_name]}", diff_bytes(diff))
+
+
+def stored_diff(folder: Path, kind_name: str, item_id: str) -> dict:
+    data = read_json(folder / f"{item_id}{DIFF_SUFFIX[kind_name]}") if kind_name in DIFF_SUFFIX else None
+    if not isinstance(data, dict) or not isinstance(data.get("diff"), list):
+        raise Missing(item_id)
+    return {"kind": kind_name, "id": item_id, "diff": data["diff"]}

orchestrator/kvb_orchestrator/fields.py

@@ -5,14 +5,15 @@ import math
 import re
 
 from .errors import ToolError
+from .signed import limits
 
 WAIT_CAP = 300
 TEXT_LIMIT = 20_000
 COMMAND_LIMIT = 4000
-PATH_LIMIT = 4096
+PATH_LIMIT = limits.PATH_LIMIT
 NAME_LIMIT = 120
 COUNT = re.compile(r"[0-9]{1,12}")
-PATH = re.compile(r"/[^\x00-\x1f\x7f]{0,4095}")
+PATH = re.compile(rf"/[^\x00-\x1f\x7f]{{0,{PATH_LIMIT - 1}}}")
 LINE = r"(?=[^\x00-\x1f\x7f]*\S)[^\x00-\x1f\x7f]{{1,{0}}}"
 ORDER_NAME = re.compile(r"(?![\s-])[^,\x00-\x1f\x7f]{1,200}")
 FIELDS = {

orchestrator/kvb_orchestrator/handler.py

@@ -6,6 +6,7 @@ from collections.abc import Callable
 from http import HTTPStatus
 from urllib.parse import urlsplit
 
+from .diffs import DIFF_BODY_LIMIT
 from .engines import STT_LIMIT
 from .exchange import PLAIN_CSP, Exchange
 
@@ -56,6 +57,7 @@ BOARD_SIGN_ROUTE = "/api/board/sign"
 SETTINGS_SIGN_ROUTE = "/api/settings/sign"
 CHOICES_ROUTE = "/api/choices"
 CHOICES_SIGN_ROUTE = "/api/choices/sign"
+DIFF_ROUTE = re.compile(r"/api/diffs/(review|ask)/([0-9a-f]{16})")
 WAV_POSTS = (STT_ROUTE, VOICE_ROUTE)
 ACTION_POSTS = ("/api/rules", "/api/nl", SETTINGS_ROUTE, TTS_ROUTE, STT_ROUTE, SIGN_ROUTE, RULE_SIGN_ROUTE, CLEANUP_ROUTE, OFFICES_ROUTE, TASKS_ROUTE, VOICE_ROUTE, MESSAGE_SIGN_ROUTE, SETTINGS_SIGN_ROUTE, CHOICES_ROUTE, CHOICES_SIGN_ROUTE)
 
@@ -83,10 +85,13 @@ class Handler(Exchange):
         cleanup = CLEANUP_STATE.fullmatch(path)
         office = OFFICE_STATE.fullmatch(path)
         voice = VOICE_ITEM.fullmatch(path)
+        diff = DIFF_ROUTE.fullmatch(path)
         if self.host() is None:
             self.send_text(HTTPStatus.FORBIDDEN, "forbidden host")
         elif path == "/":
             self.index()
+        elif diff:
+            self.answer(lambda: hub.diff_of(diff.group(1), diff.group(2)), HTTPStatus.OK)
         elif vendored and vendored.group(1) in VENDOR:
             self.vendored(*VENDOR[vendored.group(1)])
         elif voice and voice.group(2):
@@ -237,7 +242,7 @@ class Handler(Exchange):
 
     def do_POST(self) -> None:
         path = urlsplit(self.path).path
-        length = self.body_length({UPLOADS: UPLOAD_LIMIT, STT_ROUTE: STT_LIMIT, VOICE_ROUTE: STT_LIMIT}.get(path, BODY_LIMIT))
+        length = self.body_length({UPLOADS: UPLOAD_LIMIT, STT_ROUTE: STT_LIMIT, VOICE_ROUTE: STT_LIMIT, "/mcp": DIFF_BODY_LIMIT, "/api/asks": DIFF_BODY_LIMIT}.get(path, BODY_LIMIT))
         if length is None:
             return
         host = self.host()

orchestrator/kvb_orchestrator/inbox.py

@@ -19,16 +19,17 @@ SNOOZE_LIMIT = 8 * 86400
 class Inbox:
     def inbox(self, now: float, tasks: dict[str, dict], directory: Directory) -> list[dict]:
         claims = read_claims(self.council)
-        base = {"agent": "", "to": "", "detail": "", "green": False, "gate": "", "score": "", "warnings": [], "options": [], "recommended": "", "command": "", "directory": "", "sha256": "", "runnable": False, "deniable": False, "job": None, "rule": None, "dry_run": None}
+        base = {"agent": "", "to": "", "detail": "", "green": False, "gate": "", "score": "", "warnings": [], "options": [], "recommended": "", "command": "", "directory": "", "sha256": "", "runnable": False, "deniable": False, "job": None, "rule": None, "dry_run": None, "diff": None}
         items = []
         for review in self.reviews.pending():
             created = iso_epoch(review["created"]) or now
             gate = review.get("gate", "")
-            fields = {"title": review["title"], "detail": review["branch"], "green": gate == "pass", "gate": gate, "score": review.get("score", ""), "directory": review["worktree"]}
+            agent = review.get("agent", "")
+            fields = {"title": review["title"], "detail": review["branch"], "green": gate == "pass", "gate": gate, "score": review.get("score", ""), "directory": review["worktree"], "agent": agent, "to": requester_of(claims, agent)[0], "diff": review.get("diff_files")}
             items.append({**base, "kind": "review", "id": review["id"], "ts": created, **fields})
         for ask in (ask for ask in self.asks.pending() if ask["expires"] > now):
             warnings = signing.warnings(ask["command"])
-            fields = {"title": f"{ask['tool']}: {ask['command'][:200]}", "agent": directory.now(ask["session"]), "to": requester_of(claims, ask["session"])[0], "green": not warnings, "warnings": warnings, "deniable": True}
+            fields = {"title": f"{ask['tool']}: {ask['command'][:200]}", "agent": directory.now(ask["session"]), "to": requester_of(claims, ask["session"])[0], "green": not warnings, "warnings": warnings, "deniable": True, "diff": ask.get("diff_files")}
             items.append({**base, "kind": "ask", "id": ask["id"], "ts": ask["created_ts"], **fields, "command": ask["command"], "directory": ask["cwd"]})
         for decision in self.decisions.pending():
             recommended = decision.get("recommended", "")

orchestrator/kvb_orchestrator/ledger.py

@@ -110,11 +110,17 @@ class Reviews(Ledger):
     def stored(self, review_id: str, suffix: str) -> Path:
         return self.pages / f"{review_id}{suffix}"
 
-    def submit(self, fields: dict[str, str], page: bytes, files: list[dict]) -> dict:
+    def submit(self, fields: dict[str, str], page: bytes, files: list[dict], attached: dict[str, bytes]) -> dict:
+        """Args:
+            attached: further files written before the review shows, by
+                suffix.
+        """
         with self.changed:
             review_id = self.fresh()
             write_atomic(self.stored(review_id, ".html"), page)
             write_atomic(self.stored(review_id, ".json"), json.dumps({"files": files}).encode("utf-8"))
+            for suffix, data in attached.items():
+                write_atomic(self.stored(review_id, suffix), data)
             return self.insert(review_id, {**fields, "files": len(files), "status": "pending", "decided": None})
 
 

orchestrator/kvb_orchestrator/reviewing.py

@@ -6,16 +6,18 @@ import stat
 import time
 from pathlib import Path
 
+from .diffs import DIFF_SUFFIX, diff_bytes, stored_diff, valid_diff
 from .errors import Conflict, Missing, ToolError
 from .fields import WAIT_CAP, bounded, decode, expect, field, seconds
 from .review_page import review_files
-from .signed import review_signed
+from .signed import limits, review_signed
 
 PAGE_LIMIT = 32 << 20
-NOTES_LIMIT = 100_000
+NOTES_LIMIT = limits.NOTES_LIMIT
 SCORE_LIMIT = 500
 GATES = ("pass", "fail")
 REQUEST = ("repo", "worktree", "branch", "tree", "title", "page", "notes", "gate", "score")
+OPTIONAL = ("agent", "diff")
 LISTED = ("id", "created", "worktree", "branch", "tree", "title")
 STATE = ("id", "status", "tree", "decided")
 
@@ -55,7 +57,7 @@ class Reviewing:
         return data
 
     def request(self, arguments: dict) -> dict:
-        expect(arguments, REQUEST)
+        expect(arguments, (*REQUEST, *(name for name in OPTIONAL if name in arguments)))
         fields = {name: field(arguments, name) for name in ("repo", "worktree", "branch", "tree", "title")}
         notes = arguments["notes"]
         if not isinstance(notes, str) or len(notes) > NOTES_LIMIT:
@@ -63,12 +65,18 @@ class Reviewing:
         if arguments["gate"] not in GATES:
             raise ToolError("gate must be pass or fail")
         score = " ".join(bounded(arguments, "score", SCORE_LIMIT).split())
+        agent = field(arguments, "agent").strip() if "agent" in arguments else ""
+        diff = valid_diff(arguments["diff"]) if "diff" in arguments else None
         page = self.read_page(field(arguments, "page"))
-        extra = {"notes": notes, "gate": arguments["gate"], "score": score}
-        review = self.reviews.submit({**fields, "title": fields["title"].strip(), **extra}, page, review_files(page))
+        extra = {"notes": notes, "gate": arguments["gate"], "score": score, "agent": agent, "diff_files": None if diff is None else len(diff)}
+        attached = {} if diff is None else {DIFF_SUFFIX["review"]: diff_bytes(diff)}
+        review = self.reviews.submit({**fields, "title": fields["title"].strip(), **extra}, page, review_files(page), attached)
         self.automate("review", review)
         return {"id": review["id"], "status": review["status"], "url": self.url()}
 
+    def diff_of(self, kind_name: str, item_id: str) -> dict:
+        return stored_diff(self.reviews.pages, kind_name, item_id)
+
     def review_proof(self, review_id: str) -> dict:
         review = self.reviews.get(review_id)
         if review is None:

orchestrator/kvb_orchestrator/signed.py

@@ -30,6 +30,7 @@ def client_module(name: str, uses: dict[str, ModuleType]) -> ModuleType:
     return module
 
 
+limits = client_module("hub_limits", {})
 signing = client_module("signing", {})
 passkey = client_module("passkey", {"signing": signing})
 verifier = client_module("verifier", {"signing": signing, "passkey": passkey})

orchestrator/kvb_orchestrator/view.py

@@ -51,7 +51,7 @@ INDEX_CSP = (
 
 def badges(inbox: list[dict], claims: list[dict], jobs: list[dict], now: float) -> dict[str, int]:
     return {
-        "inbox": sum(1 for item in inbox if not item["until"]),
+        "tasks": sum(1 for item in inbox if not item["until"]),
         "agents": len({claim["agent"] for claim in claims if claim["beat"] is not None and claim["beat"] >= now - ACTIVE}),
         "artifacts": sum(1 for job in jobs if job["status"] in JOB_OPEN),
     }