Keep Gitea assignee aligned with agent handoffs

.gitignore

@@ -0,0 +1,3 @@
+__pycache__/
+*.py[cod]
+.pytest_cache/

scripts/lab_loop.py

@@ -68,11 +68,16 @@ class Gitea:
         if missing:
             raise LoopError(f"Missing repository labels: {', '.join(missing)}")
         self.request(
-            "PATCH",
+            "PUT",
-            f"/repos/{self.repo}/issues/{number}",
+            f"/repos/{self.repo}/issues/{number}/labels",
             {"labels": [label_map[name] for name in sorted(names)]},
         )
 
+    def set_assignee(self, number: int, agent: str) -> None:
+        if agent not in AGENTS:
+            raise LoopError(f"Unsupported assignee: {agent}")
+        self.request("PATCH", f"/repos/{self.repo}/issues/{number}", {"assignee": agent})
+
     def comment(self, number: int, body: str) -> None:
         if not body.strip():
             raise LoopError("Refusing to post an empty comment")
@@ -163,6 +168,7 @@ def cmd_handoff(api: Gitea, agent: str, args: argparse.Namespace) -> int:
     body = read_body(args, f"Continue the bounded task in issue #{args.number}.")
     api.comment(args.number, f"[HANDOFF] from={agent} to={args.to}\n\n{body.strip()}")
     api.set_labels(args.number, transition(names, agent=args.to, state="ready"))
+    api.set_assignee(args.number, args.to)
     verified = api.issue(args.number)
     validate_owner(verified, args.to, "ready")
     print(json.dumps({"status": "handed_off", "number": args.number, "from": agent, "to": args.to}))

scripts/open_daily_cycle.py

@@ -0,0 +1,102 @@
+#!/usr/bin/env python3
+"""Open exactly one Eastern-time Daily Slop Drop cycle on the recurring issue."""
+
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import json
+import os
+import urllib.error
+import urllib.request
+from zoneinfo import ZoneInfo
+
+BASE = "https://forge.alexanderwhitestone.com/git/api/v1"
+REPO = "stackchain/stackchain-lab-loop"
+TITLE = "[Recurring] Daily AI Slop Drop collaboration"
+
+
+def request(method: str, path: str, token: str, payload=None):
+    data = None if payload is None else json.dumps(payload).encode()
+    req = urllib.request.Request(
+        BASE + path,
+        data=data,
+        method=method,
+        headers={"Authorization": f"token {token}", "Content-Type": "application/json", "Accept": "application/json"},
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=20) as response:
+            raw = response.read()
+            return json.loads(raw) if raw else None
+    except urllib.error.HTTPError as exc:
+        raise RuntimeError(f"Gitea {method} failed with HTTP {exc.code}: {exc.read().decode()}") from exc
+
+
+def cycle_marker(day: dt.date) -> str:
+    return f"[CYCLE {day.isoformat()}]"
+
+
+def cycle_body(day: dt.date) -> str:
+    marker = cycle_marker(day)
+    return f"""{marker}
+
+**Timmy pass:** propose and build one concrete experience that improves on the prior drop.
+
+**Vincent pass:** challenge the concept, remove slop, and contribute one material improvement before final review.
+
+**Required audience package:**
+- a polished MP4 walkthrough that plays directly in chat;
+- the primary still or poster;
+- any interactive/source artifact;
+- verification evidence and one sharp discussion question.
+
+Post only decisions, evidence, artifacts, blockers, or explicit handoffs. No progress updates or acknowledgments."""
+
+
+def find_issue(token: str):
+    rows = request("GET", f"/repos/{REPO}/issues?state=all&type=issues&limit=100", token)
+    for row in rows:
+        if row.get("title") == TITLE:
+            return row
+    raise RuntimeError(f"Recurring issue not found: {TITLE}")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--date", help="Eastern date override, YYYY-MM-DD")
+    args = parser.parse_args()
+    token = os.environ.get("GITEA_TOKEN", "")
+    if not token:
+        raise RuntimeError("GITEA_TOKEN is required")
+    day = dt.date.fromisoformat(args.date) if args.date else dt.datetime.now(ZoneInfo("America/New_York")).date()
+    issue = find_issue(token)
+    comments = request("GET", f"/repos/{REPO}/issues/{issue['number']}/comments?limit=100", token)
+    marker = cycle_marker(day)
+    if any(marker in (comment.get("body") or "") for comment in comments):
+        print(json.dumps({"status": "already_open", "date": day.isoformat(), "issue": issue["number"]}))
+        return 0
+    labels = request("GET", f"/repos/{REPO}/labels?limit=100", token)
+    label_ids = {label["name"]: label["id"] for label in labels}
+    preserve = {
+        label["name"] for label in issue.get("labels", [])
+        if not label["name"].startswith("agent:") and not label["name"].startswith("state:")
+    }
+    desired = preserve | {"agent:timmy", "state:ready", "series:daily-slop", "cadence:daily"}
+    missing = desired - label_ids.keys()
+    if missing:
+        raise RuntimeError(f"Missing labels: {sorted(missing)}")
+    request("POST", f"/repos/{REPO}/issues/{issue['number']}/comments", token, {"body": cycle_body(day)})
+    request("PATCH", f"/repos/{REPO}/issues/{issue['number']}", token, {
+        "labels": [label_ids[name] for name in sorted(desired)],
+        "assignee": "timmy",
+    })
+    verified = request("GET", f"/repos/{REPO}/issues/{issue['number']}", token)
+    names = {label["name"] for label in verified.get("labels", [])}
+    if not {"agent:timmy", "state:ready", "series:daily-slop", "cadence:daily"}.issubset(names):
+        raise RuntimeError("Cycle label verification failed")
+    print(json.dumps({"status": "opened", "date": day.isoformat(), "issue": issue["number"], "url": issue["html_url"]}))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_daily_cycle.py

@@ -0,0 +1,20 @@
+import datetime as dt
+import unittest
+
+from scripts.open_daily_cycle import cycle_body, cycle_marker
+
+
+class DailyCycleTests(unittest.TestCase):
+    def test_marker_is_stable(self):
+        self.assertEqual(cycle_marker(dt.date(2026, 8, 7)), "[CYCLE 2026-08-07]")
+
+    def test_cycle_requires_video_and_both_agents(self):
+        body = cycle_body(dt.date(2026, 8, 7))
+        self.assertIn("Timmy pass", body)
+        self.assertIn("Vincent pass", body)
+        self.assertIn("MP4 walkthrough", body)
+        self.assertIn("No progress updates", body)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_lab_loop.py

@@ -1,9 +1,35 @@
 import unittest
 
-from scripts.lab_loop import LoopError, issue_label_names, transition, validate_owner
+from scripts.lab_loop import Gitea, LoopError, issue_label_names, transition, validate_owner
+
+
+class RecordingGitea(Gitea):
+    def __init__(self):
+        super().__init__("https://example.invalid/api/v1", "owner/repo", "secret")
+        self.calls = []
+
+    def labels(self):
+        return {"agent:timmy": 1, "state:claimed": 2, "priority:P0": 3}
+
+    def request(self, method, path, payload=None):
+        self.calls.append((method, path, payload))
 
 
 class LoopProtocolTests(unittest.TestCase):
+    def test_set_labels_uses_gitea_replace_labels_endpoint(self):
+        api = RecordingGitea()
+        api.set_labels(7, {"agent:timmy", "state:claimed", "priority:P0"})
+        self.assertEqual(
+            api.calls,
+            [
+                (
+                    "PUT",
+                    "/repos/owner/repo/issues/7/labels",
+                    {"labels": [1, 3, 2]},
+                )
+            ],
+        )
+
     def test_transition_replaces_agent_and_state_but_preserves_kind_and_priority(self):
         labels = {"agent:timmy", "state:claimed", "priority:P1", "kind:build"}
         self.assertEqual(