Create Stackchain Lab agent collaboration loop

.gitea/issue_template/task.md

@@ -0,0 +1,24 @@
+---
+name: Lab Loop Task
+title: "[Task] "
+about: A bounded Timmy/Vincent collaboration ticket
+labels: "state:ready,priority:P1"
+---
+
+## Outcome
+
+What must exist or be decided?
+
+## Owner
+
+Choose exactly one: `agent:timmy` or `agent:vincent`.
+
+## Acceptance evidence
+
+- [ ] Artifact, commit, source, or test result is linked
+- [ ] The evidence has been independently verified
+- [ ] Next state is review, blocked, or an explicit single-agent handoff
+
+## Constraints
+
+List privacy, hardware, licensing, time, or deployment limits.

AGENTS.md

@@ -0,0 +1,13 @@
+# Agent instructions
+
+Follow `PROTOCOL.md`.
+
+- Identify yourself through `LAB_AGENT`; supported initial identities are `timmy` and `vincent`.
+- Process at most one ready issue per cycle.
+- Do not post status updates, greetings, thanks, or automated acknowledgments.
+- Claim before touching work.
+- Return verified artifacts, decisions, or a single blocker.
+- Handoff to exactly one agent. Never create reciprocal automatic handoffs.
+- Do not close issues automatically; move completed work to `state:review`.
+- Keep tokens in the host environment under `GITEA_TOKEN`.
+- Never force-push or mutate another agent's active branch.

HOST_SETUP.md

@@ -0,0 +1,32 @@
+# Independent host connection
+
+Each Hermes host uses its own Forge account and scoped token. Transfer tokens privately; never place them in Telegram groups or this repository.
+
+```bash
+git clone https://forge.alexanderwhitestone.com/git/stackchain/stackchain-lab-loop.git
+cd stackchain-lab-loop
+export LAB_AGENT=vincent               # or timmy
+export GITEA_TOKEN='<host-local secret>'
+python3 scripts/lab_loop.py next
+```
+
+A worker cycle processes no more than one issue:
+
+1. Run `next`.
+2. If the result is `idle`, exit silently.
+3. Claim the returned issue:
+   ```bash
+   python3 scripts/lab_loop.py claim ISSUE_NUMBER
+   ```
+4. Read the complete issue and comments, perform the bounded work, and verify the result.
+5. Finish for human review:
+   ```bash
+   python3 scripts/lab_loop.py finish ISSUE_NUMBER --file /tmp/review-comment.md
+   ```
+   or hand off once:
+   ```bash
+   python3 scripts/lab_loop.py handoff ISSUE_NUMBER --to timmy --file /tmp/handoff.md
+   ```
+6. Do not post Telegram status. Gitea holds the durable conversation.
+
+Recommended Hermes cron behavior: run every 20–30 minutes, work only `state:ready` issues assigned to this agent, and deliver locally unless an issue carries `human-gate`.

PROTOCOL.md

@@ -0,0 +1,52 @@
+# Stackchain Lab Loop
+
+This repository is the durable conversation between independently hosted agents.
+
+## Rules
+
+1. Gitea issues are the queue and issue comments are the thread.
+2. An agent acts only on an open issue carrying both `agent:<name>` and `state:ready`.
+3. Claim before work. A claim changes `state:ready` to `state:claimed` and leaves one `[CLAIM]` receipt.
+4. Post only substantive outputs: a decision, evidence, artifact, blocker, or handoff. Never post progress chatter or acknowledgments.
+5. A handoff names exactly one next agent and changes the issue back to `state:ready`.
+6. Finished work enters `state:review`. Humans approve closure or request another handoff.
+7. One issue per concern. Branches use `<agent>/<issue>-<slug>`; no force pushes.
+8. Claims must include reproducible evidence. Links must be fetched before being reported as working.
+9. Tokens stay on each agent's host. Never commit credentials or paste them into issue threads.
+10. Three consecutive failures move the issue to `state:blocked` for human review.
+
+## Labels
+
+- `agent:timmy`, `agent:vincent`: exclusive next owner
+- `state:ready`, `state:claimed`, `state:review`, `state:blocked`: lifecycle
+- `priority:P0`, `priority:P1`, `priority:P2`: urgency
+- `kind:build`, `kind:research`, `kind:creative`, `kind:ops`: work type
+- `human-gate`: explicit Alexander/operator decision required
+
+## Comment forms
+
+```text
+[CLAIM] agent=timmy
+
+[DECISION]
+<decision and reason>
+
+[EVIDENCE]
+<verified artifact, test output, commit, or source>
+
+[HANDOFF] to=vincent
+<one concrete question or task>
+
+[BLOCKED]
+<one blocker and the exact human action required>
+```
+
+## Loop
+
+`READY → CLAIMED → REVIEW`
+
+or
+
+`READY → CLAIMED → HANDOFF(new agent) → READY`
+
+The agent utility is `scripts/lab_loop.py`. Run `python3 scripts/lab_loop.py --help` for commands.

pyproject.toml

@@ -0,0 +1,7 @@
+[project]
+name = "stackchain-lab-loop"
+version = "0.1.0"
+requires-python = ">=3.11"
+
+[tool.pytest.ini_options]
+pythonpath = ["."]

scripts/lab_loop.py

@@ -0,0 +1,242 @@
+#!/usr/bin/env python3
+"""Small, dependency-free client for the Stackchain Lab Gitea issue loop."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+DEFAULT_BASE = "https://forge.alexanderwhitestone.com/git/api/v1"
+DEFAULT_REPO = "stackchain/stackchain-lab-loop"
+AGENTS = {"timmy", "vincent"}
+STATES = {"ready", "claimed", "review", "blocked"}
+
+
+class LoopError(RuntimeError):
+    pass
+
+
+class Gitea:
+    def __init__(self, base: str, repo: str, token: str):
+        self.base = base.rstrip("/")
+        self.repo = repo.strip("/")
+        self.token = token
+
+    def request(self, method: str, path: str, payload: Any | None = None) -> Any:
+        data = None if payload is None else json.dumps(payload).encode()
+        req = urllib.request.Request(
+            f"{self.base}{path}",
+            data=data,
+            method=method,
+            headers={
+                "Accept": "application/json",
+                "Content-Type": "application/json",
+                "Authorization": f"token {self.token}",
+                "User-Agent": "stackchain-lab-loop/0.1",
+            },
+        )
+        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:
+            detail = exc.read().decode(errors="replace")
+            raise LoopError(f"Gitea {method} {path} failed: HTTP {exc.code}: {detail}") from exc
+        except urllib.error.URLError as exc:
+            raise LoopError(f"Gitea {method} {path} failed: {exc.reason}") from exc
+
+    def labels(self) -> dict[str, int]:
+        rows = self.request("GET", f"/repos/{self.repo}/labels?limit=100")
+        return {row["name"]: int(row["id"]) for row in rows}
+
+    def issue(self, number: int) -> dict[str, Any]:
+        return self.request("GET", f"/repos/{self.repo}/issues/{number}")
+
+    def issues(self) -> list[dict[str, Any]]:
+        return self.request("GET", f"/repos/{self.repo}/issues?state=open&type=issues&limit=100")
+
+    def set_labels(self, number: int, names: set[str]) -> None:
+        label_map = self.labels()
+        missing = sorted(names - label_map.keys())
+        if missing:
+            raise LoopError(f"Missing repository labels: {', '.join(missing)}")
+        self.request(
+            "PATCH",
+            f"/repos/{self.repo}/issues/{number}",
+            {"labels": [label_map[name] for name in sorted(names)]},
+        )
+
+    def comment(self, number: int, body: str) -> None:
+        if not body.strip():
+            raise LoopError("Refusing to post an empty comment")
+        self.request("POST", f"/repos/{self.repo}/issues/{number}/comments", {"body": body.strip()})
+
+
+def issue_label_names(issue: dict[str, Any]) -> set[str]:
+    return {label["name"] for label in issue.get("labels", [])}
+
+
+def transition(names: set[str], *, agent: str | None = None, state: str | None = None) -> set[str]:
+    result = {name for name in names if not name.startswith("agent:") and not name.startswith("state:")}
+    if agent:
+        if agent not in AGENTS:
+            raise LoopError(f"Unsupported agent: {agent}")
+        result.add(f"agent:{agent}")
+    if state:
+        if state not in STATES:
+            raise LoopError(f"Unsupported state: {state}")
+        result.add(f"state:{state}")
+    return result
+
+
+def validate_owner(issue: dict[str, Any], agent: str, required_state: str) -> set[str]:
+    names = issue_label_names(issue)
+    if f"agent:{agent}" not in names:
+        raise LoopError(f"Issue #{issue['number']} is not owned by agent:{agent}")
+    if f"state:{required_state}" not in names:
+        raise LoopError(f"Issue #{issue['number']} is not state:{required_state}")
+    return names
+
+
+def read_body(args: argparse.Namespace, default: str = "") -> str:
+    if getattr(args, "file", None):
+        return Path(args.file).read_text()
+    if getattr(args, "body", None):
+        return args.body
+    return default
+
+
+def cmd_next(api: Gitea, agent: str, _args: argparse.Namespace) -> int:
+    candidates = []
+    for issue in api.issues():
+        labels = issue_label_names(issue)
+        if f"agent:{agent}" in labels and "state:ready" in labels:
+            candidates.append(issue)
+    candidates.sort(key=lambda row: int(row["number"]))
+    if not candidates:
+        print(json.dumps({"status": "idle", "agent": agent}))
+        return 3
+    issue = candidates[0]
+    print(json.dumps({
+        "status": "ready",
+        "agent": agent,
+        "number": issue["number"],
+        "title": issue["title"],
+        "url": issue["html_url"],
+        "labels": sorted(issue_label_names(issue)),
+        "body": issue.get("body") or "",
+    }, indent=2))
+    return 0
+
+
+def cmd_claim(api: Gitea, agent: str, args: argparse.Namespace) -> int:
+    issue = api.issue(args.number)
+    names = validate_owner(issue, agent, "ready")
+    api.set_labels(args.number, transition(names, agent=agent, state="claimed"))
+    api.comment(args.number, f"[CLAIM] agent={agent}")
+    verified = api.issue(args.number)
+    validate_owner(verified, agent, "claimed")
+    print(json.dumps({"status": "claimed", "number": args.number, "agent": agent}))
+    return 0
+
+
+def cmd_comment(api: Gitea, agent: str, args: argparse.Namespace) -> int:
+    issue = api.issue(args.number)
+    validate_owner(issue, agent, "claimed")
+    api.comment(args.number, read_body(args))
+    print(json.dumps({"status": "commented", "number": args.number, "agent": agent}))
+    return 0
+
+
+def cmd_handoff(api: Gitea, agent: str, args: argparse.Namespace) -> int:
+    if args.to == agent:
+        raise LoopError("Refusing a handoff to the same agent")
+    issue = api.issue(args.number)
+    names = validate_owner(issue, agent, "claimed")
+    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"))
+    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}))
+    return 0
+
+
+def cmd_finish(api: Gitea, agent: str, args: argparse.Namespace) -> int:
+    issue = api.issue(args.number)
+    names = validate_owner(issue, agent, "claimed")
+    body = read_body(args)
+    if body.strip():
+        api.comment(args.number, f"[REVIEW] agent={agent}\n\n{body.strip()}")
+    api.set_labels(args.number, transition(names, agent=agent, state="review"))
+    verified = api.issue(args.number)
+    validate_owner(verified, agent, "review")
+    print(json.dumps({"status": "review", "number": args.number, "agent": agent}))
+    return 0
+
+
+def cmd_block(api: Gitea, agent: str, args: argparse.Namespace) -> int:
+    issue = api.issue(args.number)
+    names = validate_owner(issue, agent, "claimed")
+    body = read_body(args)
+    api.comment(args.number, f"[BLOCKED] agent={agent}\n\n{body.strip()}")
+    api.set_labels(args.number, transition(names, agent=agent, state="blocked"))
+    print(json.dumps({"status": "blocked", "number": args.number, "agent": agent}))
+    return 0
+
+
+def parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("--base", default=os.getenv("GITEA_BASE", DEFAULT_BASE))
+    p.add_argument("--repo", default=os.getenv("LAB_REPO", DEFAULT_REPO))
+    p.add_argument("--agent", default=os.getenv("LAB_AGENT", ""))
+    sub = p.add_subparsers(dest="command", required=True)
+    sub.add_parser("next")
+    for name in ("claim", "comment", "finish", "block"):
+        cmd = sub.add_parser(name)
+        cmd.add_argument("number", type=int)
+        if name != "claim":
+            source = cmd.add_mutually_exclusive_group(required=True)
+            source.add_argument("--body")
+            source.add_argument("--file")
+    handoff = sub.add_parser("handoff")
+    handoff.add_argument("number", type=int)
+    handoff.add_argument("--to", required=True, choices=sorted(AGENTS))
+    source = handoff.add_mutually_exclusive_group()
+    source.add_argument("--body")
+    source.add_argument("--file")
+    return p
+
+
+def main() -> int:
+    args = parser().parse_args()
+    if args.agent not in AGENTS:
+        raise LoopError("Set LAB_AGENT to timmy or vincent, or pass --agent")
+    token = os.getenv("GITEA_TOKEN", "")
+    if not token:
+        raise LoopError("GITEA_TOKEN is required")
+    api = Gitea(args.base, args.repo, token)
+    handlers = {
+        "next": cmd_next,
+        "claim": cmd_claim,
+        "comment": cmd_comment,
+        "handoff": cmd_handoff,
+        "finish": cmd_finish,
+        "block": cmd_block,
+    }
+    return handlers[args.command](api, args.agent, args)
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except LoopError as exc:
+        print(json.dumps({"status": "error", "error": str(exc)}), file=sys.stderr)
+        raise SystemExit(2)

tests/test_lab_loop.py

@@ -0,0 +1,38 @@
+import unittest
+
+from scripts.lab_loop import LoopError, issue_label_names, transition, validate_owner
+
+
+class LoopProtocolTests(unittest.TestCase):
+    def test_transition_replaces_agent_and_state_but_preserves_kind_and_priority(self):
+        labels = {"agent:timmy", "state:claimed", "priority:P1", "kind:build"}
+        self.assertEqual(
+            transition(labels, agent="vincent", state="ready"),
+            {"agent:vincent", "state:ready", "priority:P1", "kind:build"},
+        )
+
+    def test_same_agent_transition_is_idempotent(self):
+        labels = {"agent:timmy", "state:ready", "priority:P0"}
+        self.assertEqual(transition(labels, agent="timmy", state="ready"), labels)
+
+    def test_invalid_agent_fails_closed(self):
+        with self.assertRaises(LoopError):
+            transition(set(), agent="unknown", state="ready")
+
+    def test_validate_owner_requires_both_labels(self):
+        issue = {
+            "number": 7,
+            "labels": [{"name": "agent:vincent"}, {"name": "state:claimed"}],
+        }
+        self.assertEqual(validate_owner(issue, "vincent", "claimed"), {"agent:vincent", "state:claimed"})
+        with self.assertRaises(LoopError):
+            validate_owner(issue, "timmy", "claimed")
+        with self.assertRaises(LoopError):
+            validate_owner(issue, "vincent", "ready")
+
+    def test_issue_label_names_handles_missing_labels(self):
+        self.assertEqual(issue_label_names({}), set())
+
+
+if __name__ == "__main__":
+    unittest.main()