commit fdb454414d0a1e8f70f686bacb32c30ab6063c34 Author: Timmy Jr. Date: Fri Aug 7 13:27:37 2026 +0000 Create Stackchain Lab agent collaboration loop diff --git a/.gitea/issue_template/task.md b/.gitea/issue_template/task.md new file mode 100644 index 0000000..6927755 --- /dev/null +++ b/.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. diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..bb96cd3 --- /dev/null +++ b/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. diff --git a/HOST_SETUP.md b/HOST_SETUP.md new file mode 100644 index 0000000..1854dbe --- /dev/null +++ b/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='' +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`. diff --git a/PROTOCOL.md b/PROTOCOL.md new file mode 100644 index 0000000..b27afd8 --- /dev/null +++ b/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:` 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 `/-`; 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] + + +[EVIDENCE] + + +[HANDOFF] to=vincent + + +[BLOCKED] + +``` + +## 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. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..e9204ce --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,7 @@ +[project] +name = "stackchain-lab-loop" +version = "0.1.0" +requires-python = ">=3.11" + +[tool.pytest.ini_options] +pythonpath = ["."] diff --git a/scripts/__pycache__/lab_loop.cpython-311.pyc b/scripts/__pycache__/lab_loop.cpython-311.pyc new file mode 100644 index 0000000..15cc71a Binary files /dev/null and b/scripts/__pycache__/lab_loop.cpython-311.pyc differ diff --git a/scripts/lab_loop.py b/scripts/lab_loop.py new file mode 100644 index 0000000..0f4cfc9 --- /dev/null +++ b/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) diff --git a/tests/__pycache__/test_lab_loop.cpython-311.pyc b/tests/__pycache__/test_lab_loop.cpython-311.pyc new file mode 100644 index 0000000..eda06fa Binary files /dev/null and b/tests/__pycache__/test_lab_loop.cpython-311.pyc differ diff --git a/tests/test_lab_loop.py b/tests/test_lab_loop.py new file mode 100644 index 0000000..0216acc --- /dev/null +++ b/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()