dev/nisse
2
1 Fork 0
Code Issues 23 Pull Requests Actions Packages Projects Releases 1 Wiki Activity
Files
dd6196c742387f972553c472ad095041c9a590f4
nisse/nisse-dev/src/nisse_dev/loop/agent.py
T
dongarandClaude Opus 5.5 dd6196c742
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 4m56s
Details
New projects from conversation (M2.5, PR B) 
- propose_project(name, purpose, decisions) in a conversation goes up for
  a go-ahead (checked: a slug, not an existing project, no trailing
  question).
- On go, the provisioning checklist (D1): a repository with a README,
  protection on its default branch, the agent as a collaborator, a build
  VM, a deploy key, and the project entry as YAML with <...> to fill in.
- On "done", the agent reads its configuration again and adds the new
  project live (build_loop_agent(load=...); only additions), then checks
  what was provided (D4, verified, not assumed): permissions and
  protection as it sees them, then on a throwaway branch with an empty
  commit, a branch push must succeed and a push to the default branch and
  merging its own PR must be refused. The check's PR is closed and its
  branch deleted. Failures are said per item; nothing starts.
- Checked: the tracking issue (the distilled decisions, carrying the new
  item's ID), and the project's item, agreed and labelled `design`, whose
  first milestone is ARCHITECTURE.md and the build plan. A conversation
  waiting on provisioning doesn't close as idle.
- The walk replays exactly from its seed (item IDs too), and event IDs
  stay unique across simulated restarts, as a server's are.
- The test config holds output to 8192 tokens, under the test key's
  per-request cost cap.

Tests: new-project unit tests (checklist and its YAML parse as config, done
before the entry, what isn't set up, a checked project's record and first
PR, an existing name), protection enforced by a hook on the fake forge,
the walk proposing projects and saying done, and an e2e new project in a
repository the test operator provisions from the agent's checklist.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE
2026-10-06 14:07:37 +00:00

1695 lines
72 KiB
Python
Raw Blame History

"""The loop's scheduler (L1, L2, L5, D1): intake, messages and driver runs.
Everything durable happens on one worker thread, in order. Inputs (an
operator's message, an issue assigned to the agent) are handed to it as they
arrive; a message is stamped with the revision on screen in its thread first,
because by the time the worker reaches it a newer one may be up (L3).
One work item at a time is active. An item runs until its driver calls a
waiting tool; it then waits on the operator, the active pointer is cleared
(a safe point, L1), and the next ready item runs: the operator's order when
given ("do #14 next"), otherwise the oldest. Switching parks the item that was
on the workspace: its work is committed and its branch pushed (D4). An issue
with open Gitea dependencies is blocked until they close. Conversation is not work:
a message is recorded and changes its item's state, but never interrupts a
run (L1).
"""
import re
import threading
import time
import uuid
from collections.abc import Callable, Iterable, Sequence
from dataclasses import dataclass, replace
from nisse_dev.cancel import Cancel
from nisse_dev.canonical import Fact, PlanSubmitted, UserMessage
from nisse_dev.config import GENERAL
from nisse_dev.gates import command_form
from nisse_dev.git import GitError
from nisse_dev.gitea import GiteaClient, GiteaError, Issue
from nisse_dev.loop.driver import Driver, RunOutcome
from nisse_dev.loop.items import (
ANSWER,
BLOCKED,
CHAT,
CLOSED,
GO_AHEAD,
MERGE,
PROVISION,
READY,
RUNNING,
WAITING,
Loop,
item_for_issue,
item_for_thread,
loop_of,
save,
)
from nisse_dev.loop.milestones import OPEN_PR, Milestones
from nisse_dev.loop.reader import GoAheadReader, is_question
from nisse_dev.loop.tools import (
ASK_HUMAN,
CLOSE_ITEM,
FINISH_MILESTONE,
PROPOSE_PROJECT,
PROPOSE_WORK,
REPLY,
RESTATE,
START_MILESTONE,
WAIT_FOR_REVIEW,
)
from nisse_dev.oplog import OpLog
from nisse_dev.state import StateStore, WorkItem
from nisse_dev.tools import EXECUTOR_TOOLS, Workspace
from nisse_dev.transport import InboundMessage, Transport
from nisse_dev.worker import Worker
INTAKE_INTERVAL_S = 30.0
_GO = frozenset({"go", "approve"})
# An issue, named: "#14", "14", or with its project, "sandbox#14" (D8).
_ISSUE = r"(?:([a-z0-9][a-z0-9-]*)#|#)?(\d+)"
_NEXT = re.compile(rf"(?:do|work on) {_ISSUE} (?:next|first|now)")
# A command that names its item applies to it, wherever it is said (L5).
_NAMED = re.compile(r"(go|approve|abandon|stop) (?:([a-z0-9][a-z0-9-]*))?#(\d+)")
# At top level these need an item; without one they are ambiguous (L5).
_NEEDS_ITEM = frozenset({"go", "approve", "abandon"})
BUG_GRACE_S = 600.0
GENERAL_IDLE_S = 86400.0
FILE_ISSUE = "file_issue" # the intent of opening a record issue (L8)
# Carried by a record issue the agent opens for work started in conversation:
# intake never takes it for new work, even before its item exists (D1).
_MARKER = re.compile(r"<!-- nisse-item: ([A-Za-z0-9-]+) -->")
# Before the go-ahead the driver may look around (read, list, run a command
# to reproduce a bug) but not change anything (D1); after it, everything.
_LOOK = [t for t in EXECUTOR_TOOLS if t["function"]["name"] != "write_file"]
SYSTEM = """You are nisse-dev, a software agent working with the operator on
the repository {repo}. You are on one work item: issue #{issue}, "{title}".
The issue and the conversation about it are below.
Before any work, find out what you need, then use restate to put your
understanding to the operator: what you take the request to mean, how it fits
the project, and your first step. Wait for their go-ahead. If the request is
unclear, ask_human instead of guessing: a restatement asks no questions, so
the operator can agree to it as it stands.
The project's documents are in its repository: read ARCHITECTURE.md and any
build plans under docs/plans/ before you restate.
After the go-ahead, work in milestones, one PR each. Call start_milestone
before changing anything (never create or switch branches yourself; each
milestone starts from the latest default branch), make the change, run the
project's tests, then call finish_milestone with a summary, saying whether it
is the final milestone. The harness commits, pushes and opens a PR, and you
wait for review. Review comments arrive as messages: make the changes within
the same milestone (don't start another), run the tests, and call
finish_milestone again to push the fix. When a review needs no change from
you, or you are only waiting for the operator to merge (you can't merge it
yourself), call wait_for_review. The next milestone starts only after the
previous PR has merged. When the last PR has merged, call close_item with
what was done, what was verified, and what is left open. If a PR is closed
without merging, ask the operator what next.
Work longer than about one PR gets a build plan: a Markdown file under
docs/plans/ giving, per milestone, the goal, the interfaces and behaviour, the
edge cases, and the tests that are its acceptance criteria. Commit it with the
first milestone, and mark each milestone done in it in that milestone's PR.
Keep ARCHITECTURE.md current: a design change updates it in the same PR.
{labels}Never label the issue yourself; you may suggest a label to the operator.
{workspace}"""
CONVERSATION = """You are nisse-dev, a software agent, in a conversation with
the operator in a Matrix thread. It isn't about a project yet: it may be a
question, thoughts on feasibility or hardware, or the start of some work.
Answer with reply; ask_human only for a question you need answered. You have
no workspace and can't run anything here; if an experiment is needed, say so.
If the conversation turns into work on one of these projects, use
propose_work, with a summary of what was decided (not the conversation): it
is the new issue's body and the work's whole brief.
{projects}
If it becomes a new project, use propose_project: its name, what it's for,
and the decisions setting it up depends on. Once agreed, you'll be asked for
what it needs set up; when the operator has it ready and it checks out, its
architecture and build plan are the first piece of work."""
# What an issue's labels add to the driver's instructions (D1).
LABELS = {
"bug": "This issue is labelled bug. Before restating, reproduce it with "
"your tools (run the code or the tests). If you can't reproduce it, "
"ask_human instead of restating: never fix a bug you couldn't see. Once "
"you restate, the fix goes ahead after a grace period unless the operator "
"replies; make it a failing test first, then the fix.",
"discuss": "This issue is labelled discuss: talk it through with the "
"operator (ask_human) before restating, and restate only once you agree "
"on what is wanted.",
"design": "This issue is labelled design: agree the design with the "
"operator before any code. Your restatement puts the design to them; "
"code waits for their go-ahead on it.",
}
@dataclass(frozen=True)
class Project:
"""What the loop needs of one project (D8): its repository, label
defaults, and its own workspace and milestones on its build VM."""
name: str
repo: str
workspace: Workspace | None = None
milestones: Milestones | None = None
labels: tuple[str, ...] = ()
def _restatement_problem(args: dict[str, object]) -> str | None:
"""L5, D1: a restatement is agreed to as it stands, so it asks nothing.
One that ends in a question is sent back: questions go by ask_human."""
text = str(args.get("text", "")).strip()
if not text:
return "a restatement needs text: what you take the request to mean"
if is_question(text):
return (
"not posted: a restatement asks no questions, so the operator can "
"agree to it as it stands. If you need an answer first, use "
"ask_human; otherwise restate it as a statement."
)
return None
class LoopAgent:
def __init__(
self,
store: StateStore,
transport: Transport,
gitea: GiteaClient,
*,
account: str,
operators: Iterable[str],
gitea_operators: Iterable[str],
driver: Driver,
worker: Worker,
cancel: Cancel,
oplog: OpLog,
projects: Sequence[Project] = (),
# One project, given directly: the same as projects=[Project(...)].
repo: str = "",
project: str = "",
workspace: Workspace | None = None,
roles: dict[str, str] | None = None,
milestones: Milestones | None = None,
reader: GoAheadReader | None = None,
bug_grace_s: float = BUG_GRACE_S,
general_idle_s: float = GENERAL_IDLE_S,
# A new project's entry, once the operator has added it (D1, D8):
# the projects configured now that this agent doesn't know yet.
reload: Callable[[set[str]], Sequence[Project]] | None = None,
clock: Callable[[], float] = time.time,
intake_interval_s: float = INTAKE_INTERVAL_S,
) -> None:
self._store = store
self._transport = transport
self._gitea = gitea
if not projects:
projects = [Project(project, repo, workspace, milestones)]
self._projects = {p.name: p for p in projects}
self._account = account
self._operators = frozenset(operators)
self._gitea_operators = frozenset(gitea_operators)
self._driver = driver
self._worker = worker
self._cancel = cancel
self._oplog = oplog
self._roles = roles or {}
self._reader = reader
self._grace = bug_grace_s
self._idle = general_idle_s
self._reload = reload
self._clock = clock
self._interval = intake_interval_s
# thread root -> revision on screen there, read on the transport's
# thread to stamp replies on arrival (L3); (project, issue) -> thread
# root, for a command that names its item.
self._shown: dict[str, int] = {}
self._threads: dict[tuple[str, int], str] = {}
self._active_thread = ""
# The project whose workspace was last worked on: switching to
# another parks it (L1). None after a boot: unknown, so all are.
self._last_project: str | None = None
self._running = False
self._stop = threading.Event()
self._intake_thread: threading.Thread | None = None
# -- projects (D8) ----------------------------------------------------------
def _at(self, item: WorkItem) -> Project:
if item.scope == GENERAL:
# A conversation: no repository, workspace or build VM (L1).
return Project(GENERAL, "")
return self._projects[item.scope or ""]
def _label(self, item: WorkItem) -> str:
"""How an item's issue is named to the operator: #14 with one
project, sandbox#14 with more, where #14 alone could be either."""
if item.scope == GENERAL:
return f'the conversation "{item.title}"'
issue = loop_of(item).issue
return f"#{issue}" if len(self._projects) == 1 else f"{item.scope}#{issue}"
def _named(self, project: str | None, issue: int) -> list[WorkItem]:
"""Open items for ``issue``, in ``project`` if it is given."""
return [
i
for i in self._store.work_items("open")
if i.scope != GENERAL
and loop_of(i).issue == issue
and (project is None or i.scope == project)
]
def _repos(self) -> str:
return ", ".join(p.repo for p in self._projects.values())
# -- lifecycle ----------------------------------------------------------
def start(self) -> None:
self._worker.call(self.boot)
self._transport.on_message(self.on_message)
start = getattr(self._transport, "start", None)
if start is not None:
start()
self._intake_thread = threading.Thread(target=self._intake_loop, daemon=True)
self._intake_thread.start()
def stop(self) -> None:
self._stop.set()
if self._intake_thread is not None:
self._intake_thread.join(timeout=5)
def _intake_loop(self) -> None:
while not self._stop.is_set():
self._worker.submit(self._settled("intake", self.intake))
self._stop.wait(self._interval)
def _settled(self, what: str, job: Callable[[], None]) -> Callable[[], None]:
"""A worker job that reports a failure instead of losing it."""
def run() -> None:
try:
job()
except Exception as exc: # noqa: BLE001 -- reported, the loop goes on
self._oplog.emit("loop.job_failed", cause=what, error=repr(exc))
self._transport.post(f"something went wrong with {what}: {exc}")
return run
# -- boot (L8) --------------------------------------------------------------
def boot(self) -> None:
"""Resume waiting points as they were, and never resume interrupted
work blindly: an item cut off mid-run waits on the operator."""
self._store.set_active(None, cause="boot")
self._reconcile_publishing()
self._reconcile_filing()
for item in self._store.work_items("open"):
loop = loop_of(item)
self._threads[(item.scope or "", loop.issue)] = loop.thread
if item.position == RUNNING:
note = "I restarted while working on this. Reply to continue."
self._wait(item, replace(loop, wait=ANSWER, note=note), note)
elif item.position == WAITING and loop.thread:
if loop.wait == GO_AHEAD:
self._shown[loop.thread] = loop.rev
self._transport.post(
f"(after a restart) rev-{loop.rev} is still up; say go "
"to start, or reply with changes.",
thread=loop.thread,
)
elif loop.wait != CHAT: # a conversation just goes on
self._transport.post(
f"(after a restart) still waiting: {loop.note}",
thread=loop.thread,
)
def _reconcile_filing(self) -> None:
"""L8: a record issue cut off by a crash. An issue carrying the new
item's ID was opened: its item is started now. Otherwise nothing was
opened, and the conversation's proposal is up again."""
for intent in self._store.pending_intents():
if intent.kind != FILE_ISSUE or not intent.work_item:
continue
conversation = self._store.work_item(intent.work_item)
repo = str(intent.payload.get("repo", ""))
new_id = str(intent.payload.get("item", ""))
project = next((p for p in self._projects.values() if p.repo == repo), None)
if self._store.work_item(new_id) is not None:
self._store.commit_intent(intent.id, cause="its item exists")
continue
if conversation is None or project is None:
self._store.abort_intent(
intent.id, cause="conversation or project gone"
)
continue
try:
issues = self._gitea.assigned_issues(repo, self._account)
except GiteaError:
continue # unverifiable now; settled at the next boot
filed = [
i
for i in issues
if (m := _MARKER.search(i.body or "")) and m.group(1) == new_id
]
if filed:
self._start_filed(
conversation,
project,
new_id,
filed[0].number,
filed[0].title,
self._latest_proposal(conversation),
)
self._store.commit_intent(
intent.id, cause=f"adopted {repo}#{filed[0].number}"
)
else:
self._store.abort_intent(intent.id, cause="no issue was opened")
loop = loop_of(conversation)
note = (
"opening the issue was cut off by a restart; say go to try again."
)
self._shown[loop.thread] = loop.rev
self._wait(conversation, replace(loop, wait=GO_AHEAD, note=note), note)
def _reconcile_publishing(self) -> None:
"""L8 step 2: a publish cut off by a crash is settled against Gitea.
An open PR for the branch is adopted; otherwise the intent is aborted
and the operator is told, never a blind retry."""
for intent in self._store.pending_intents():
if intent.kind != OPEN_PR or not intent.work_item:
continue
item = self._store.work_item(intent.work_item)
branch = str(intent.payload.get("branch", ""))
if item is None or item.scope not in self._projects:
self._store.abort_intent(intent.id, cause="item or project unknown")
continue
try:
pr = self._gitea.find_pull_request(self._at(item).repo, branch)
except GiteaError:
continue # unverifiable: left pending, asked about below
if item is None or item.status != "open":
self._store.abort_intent(intent.id, cause="item no longer open")
elif pr is not None and pr.state == "open":
self._store.commit_intent(intent.id, cause=f"adopted PR #{pr.number}")
loop = replace(loop_of(item), pr=pr.number, pr_url=pr.url)
self._fact(item, f"after a restart, adopted PR #{pr.number} ({pr.url})")
save(
self._store,
item.id,
WAITING,
replace(loop, wait=MERGE, note=f"PR #{pr.number}"),
cause="adopted PR after restart",
)
else:
self._store.abort_intent(intent.id, cause="no PR was opened")
note = "publishing was cut off by a restart; reply to try again."
self._wait(item, replace(loop_of(item), wait=ANSWER, note=note), note)
# -- intake (D1) ---------------------------------------------------------
def intake(self) -> None:
"""Issues assigned to the agent, in every project's repository,
become work items, once each."""
for project in self._projects.values():
self._intake(project)
self.watch()
self.check_blocked()
self.check_grace()
self.check_idle()
self._schedule()
def _intake(self, project: Project) -> None:
try:
issues = self._gitea.assigned_issues(project.repo, self._account)
except GiteaError as exc:
# L5: an outage delays intake, never fails anything.
self._oplog.emit("loop.intake_unavailable", cause=str(exc))
return
for issue in issues:
if item_for_issue(self._store, issue.number, project.name) is not None:
continue
if _MARKER.search(issue.body or ""):
continue # D1: a record of work started in conversation
if not self._from_an_operator(project, issue):
self._oplog.emit(
"loop.intake_ignored",
cause="not filed or assigned by an operator",
issue=issue.number,
repo=project.repo,
)
continue
try:
blocked_by = self._gitea.open_dependencies(project.repo, issue.number)
except GiteaError as exc:
self._oplog.emit("loop.intake_unavailable", cause=str(exc))
continue # taken on the next round
self._open(project, issue, tuple(blocked_by))
# -- PR review and merge (D4, D5) -------------------------------------------
def watch(self) -> None:
"""New operator comments on an item's PR, and its merge or close,
reach the item; a waiting item becomes ready to act on them."""
for item in self._store.work_items("open"):
loop = loop_of(item)
if not loop.pr:
continue
repo = self._at(item).repo
try:
pr = self._gitea.pull_request(repo, loop.pr)
comments = self._gitea.list_comments(repo, loop.pr)
notes = self._gitea.review_notes(repo, loop.pr)
except GiteaError as exc:
self._oplog.emit("loop.watch_unavailable", cause=str(exc))
continue
log = self._store.transcript(item.id, "operator")
new = False
for comment in comments:
if comment.id <= loop.comments:
continue
loop = replace(loop, comments=comment.id)
if comment.author in self._gitea_operators:
log.append(
UserMessage(
text=comment.body,
source=f"gitea:pr:{loop.pr}:comment:{comment.id}",
)
)
loop = replace(
loop,
asked_on_pr=loop.pr,
asked_where="",
asked_text=comment.body[:200],
)
new = True
for note in notes:
if note.key in loop.notes:
continue
loop = replace(loop, notes=(*loop.notes, note.key))
if note.author in self._gitea_operators:
where = f"on {note.where}: " if note.where else ""
log.append(
UserMessage(
text=f"{where}{note.body}",
source=f"gitea:pr:{loop.pr}:review:{note.key}",
)
)
loop = replace(
loop,
asked_on_pr=loop.pr,
asked_where=note.where,
asked_text=note.body[:200],
)
new = True
position = item.position
if pr.merged:
self._fact(item, f"PR #{loop.pr} merged")
loop, position = replace(loop, pr=0, pr_url="", branch=""), READY
elif pr.state != "open":
self._fact(item, f"PR #{loop.pr} was closed without merging")
loop, position = replace(loop, pr=0, pr_url="", branch=""), READY
elif new and item.position == WAITING:
position = READY
if position != item.position or loop != loop_of(item):
if position == READY:
loop = replace(loop, wait="", note="")
save(self._store, item.id, position, loop, cause="PR update")
def check_blocked(self) -> None:
"""L1: an item that hasn't started running yet is blocked while its
Gitea dependencies are open, and ready once they have all closed.
Only ready and blocked items are checked: the others are mid-work or
waiting on the operator, and an agent that finds a blocker mid-work
asks before recording it."""
for item in self._store.work_items("open"):
if item.position not in (READY, BLOCKED) or item.scope == GENERAL:
continue
loop = loop_of(item)
repo = self._at(item).repo
try:
blocked_by = tuple(self._gitea.open_dependencies(repo, loop.issue))
except GiteaError as exc:
self._oplog.emit("loop.watch_unavailable", cause=str(exc))
continue # as it was, until Gitea answers
position = BLOCKED if blocked_by else READY
if position == item.position and blocked_by == loop.blocked_by:
continue
name = self._label(item)
if position == READY:
said = f"{name} is unblocked: its dependencies have closed."
else:
said = f"{name} is blocked by {', '.join(blocked_by)}."
self._fact(item, said)
if position != item.position:
self._transport.post(said, thread=loop.thread)
save(
self._store,
item.id,
position,
replace(loop, blocked_by=blocked_by),
cause=said,
)
def check_idle(self) -> None:
"""L1: a conversation left without a message for a day closes; a
later reply in its thread starts a new one, linked to it."""
now = self._clock()
for item in self._store.work_items("open"):
loop = loop_of(item)
if (
item.scope == GENERAL
and item.position == WAITING
and not loop.new_project # setting one up can take days
and loop.idle_since
and now - loop.idle_since >= self._idle
):
said = sum(
1
for e in self._store.transcript(item.id, "operator").read()
if e.actor == "operator"
)
became = f"; it became {', '.join(loop.became)}" if loop.became else ""
self._close(
item,
loop,
"idle",
f"no message for {round(self._idle / 3600)} h, after {said} "
f"from the operator{became}",
)
def check_grace(self) -> None:
"""D1: a `bug` restatement left unanswered for the grace period is
taken as agreed, bound to the revision that was up all along."""
now = self._clock()
for item in self._store.work_items("open"):
loop = loop_of(item)
if (
item.position == WAITING
and loop.wait == GO_AHEAD
and loop.grace_until
and now >= loop.grace_until
):
self._transport.post(
f"No reply on rev-{loop.rev}, so I'm going ahead.",
thread=loop.thread,
)
self._go(item, loop, "the bug grace period")
def _from_an_operator(self, project: Project, issue: Issue) -> bool:
"""L11: only an operator's issue, or one an operator assigned, is
an instruction."""
if issue.author in self._gitea_operators:
return True
try:
who = self._gitea.assigned_by(project.repo, issue.number, self._account)
except GiteaError:
return False
return who in self._gitea_operators
def _open(
self, project: Project, issue: Issue, blocked_by: tuple[str, ...]
) -> None:
item_id = f"{project.name}-{issue.number}-{uuid.uuid4().hex[:6]}"
cause = f"assigned {project.repo}#{issue.number}"
ahead = len(self._store.work_items("open"))
where = f", {ahead} ahead" if ahead else ""
if blocked_by:
where += f"; blocked by {', '.join(blocked_by)}"
root = self._transport.post(
f"Queued {project.repo}#{issue.number}: {issue.title}{where}"
)
# D8: the project's label defaults steer as if the issue had them.
labels = tuple(dict.fromkeys((*issue.labels, *project.labels)))
loop = Loop(
thread=root,
issue=issue.number,
blocked_by=blocked_by,
labels=labels,
)
self._threads[(project.name, issue.number)] = root
self._store.open_work_item(
item_id,
title=issue.title,
scope=project.name,
position=BLOCKED if blocked_by else READY,
data={"loop": loop.to_data(), "roles": self._roles},
cause=cause,
)
log = self._store.transcript(item_id, "orchestrator")
log.append(
Fact(
text=f"work item opened from {project.repo}#{issue.number} "
f"({issue.title}), filed by {issue.author}"
+ (f", labels: {', '.join(labels)}" if labels else "")
+ (
f" ({', '.join(project.labels)} by the project's default)"
if project.labels
else ""
)
)
)
text = f"{issue.title}\n\n{issue.body}".strip()
if issue.author in self._gitea_operators:
log.as_actor("operator").append(
UserMessage(text=text, source=f"gitea:issue-body:{issue.number}")
)
else:
# L11: someone else's text is data; the operator's assignment is
# the instruction.
log.append(Fact(text=f"the issue as written by {issue.author}:\n{text}"))
# -- messages (L5) -------------------------------------------------------
def on_message(self, msg: InboundMessage) -> None:
"""On the transport's thread: stamp, then hand to the worker."""
if msg.sender not in self._operators:
return
command = command_form(msg.text)
thread = msg.thread
named = _NAMED.fullmatch(command)
if named is not None:
command = named.group(1)
project, issue = named.group(2), int(named.group(3))
threads = [
t
for (p, n), t in list(self._threads.items())
if n == issue and project in (None, p)
]
# Ambiguous across projects: no stamp; handle asks which.
thread = threads[0] if len(threads) == 1 else thread
if (
command == "stop"
and self._running
# Unnamed at top level, it stops the active item (L5).
and (not thread or thread == self._active_thread)
):
# Straight to the flag: the run is what it has to reach (L2).
self._cancel.set()
about = self._shown.get(thread) if thread else None
self._worker.submit(
self._settled("your message", lambda: self.handle(msg, about))
)
def handle(self, msg: InboundMessage, about: int | None) -> None:
item = item_for_thread(self._store, msg.thread) if msg.thread else None
text = (msg.original or msg.text).strip()
command = command_form(msg.text)
order = _NEXT.fullmatch(command)
if order is not None:
self._store.transcript(item.id if item else None, "operator").append(
UserMessage(text=text, source=f"transport:{msg.sender}")
)
self._transport.post(
self._put_first(order.group(1), int(order.group(2))),
thread=msg.thread,
)
self._schedule()
return
named = _NAMED.fullmatch(command)
if named is not None:
# L5: a command that names its item applies to it.
command = named.group(1)
project, issue = named.group(2), int(named.group(3))
found = self._named(project, issue)
if len(found) != 1:
self._store.transcript(None, "operator").append(
UserMessage(text=text, source=f"transport:{msg.sender}")
)
self._transport.post(
self._not_one(found, project, issue, f"{command} {{}}"),
thread=msg.thread,
)
return
[item] = found
if item is None and command not in ("status", "stop", *_NEEDS_ITEM):
# L5: any other message is a conversation, from its first message.
self._converse(msg, text)
self._schedule()
return
if item is None:
self._store.transcript(None, "operator").append(
UserMessage(text=text, source=f"transport:{msg.sender}")
)
if command == "status":
self._transport.post(self.status(), thread=msg.thread)
elif command in _NEEDS_ITEM:
self._transport.post(self._which(command), thread=msg.thread)
elif not self._store.active_item():
self._transport.post("Nothing is running.", thread=msg.thread)
return
self._store.transcript(item.id, "operator").append(
UserMessage(text=text, source=f"transport:{msg.sender}")
)
# Any reply stops a grace period's countdown: the operator is here.
loop = replace(
loop_of(item),
asked_on_pr=0,
asked_where="",
asked_text="",
grace_until=0.0,
talked=True,
idle_since=self._clock(),
)
if loop != loop_of(item):
save(self._store, item.id, item.position, loop, cause="operator spoke")
item = self._store.work_item(item.id) or item
if command == "abandon":
self._close(item, loop, "abandoned", f"abandoned by {msg.sender}")
elif command == "done" and loop.new_project:
self._provisioned(item, loop)
elif command == "stop":
# Stopping a run took effect on arrival; a stop is never an answer.
if not loop.note.startswith("stopped ("):
self._transport.post(
f"{self._label(item)} isn't running; nothing to stop.",
thread=loop.thread,
)
elif item.position == BLOCKED:
self._transport.post(
f"{self._label(item)} is blocked by {', '.join(loop.blocked_by)}; "
"I'll start on it when they close, or when you remove them.",
thread=loop.thread,
)
elif item.position == WAITING and loop.wait == GO_AHEAD and command in _GO:
if about != loop.rev:
# L3: a go-ahead is about what was on screen when it arrived.
self._transport.post(
f"that go was for rev-{about}, but rev-{loop.rev} is up now; "
"say go again if it still stands.",
thread=loop.thread,
)
return
self._go(item, loop, msg.sender)
elif (
item.position == WAITING
and loop.wait == GO_AHEAD
and about == loop.rev
and self._plain_go(item, text)
):
# L5, D1: a plain-language yes, said so before acting.
self._transport.post(
f"Taking that as a go-ahead on rev-{loop.rev}.", thread=loop.thread
)
self._go(item, loop, f"{msg.sender} (in plain language)")
elif item.position == WAITING:
# An answer, or changes to a restatement: the driver reads it.
save(
self._store,
item.id,
READY,
replace(loop, wait="", note=""),
cause="the operator answered",
)
self._schedule()
def _converse(self, msg: InboundMessage, text: str) -> None:
"""A conversation (L1): its own item in the `general` scope, in a
thread under the operator's message, or in the thread of one that
closed, which it continues."""
thread = msg.thread or msg.event_id
before = (
item_for_thread(self._store, thread, status=None) if msg.thread else None
)
title = text.splitlines()[0][:60] if text else "conversation"
item_id = f"{GENERAL}-{uuid.uuid4().hex[:8]}"
loop = Loop(
thread=thread,
idle_since=self._clock(),
talked=True,
follows=before.id if before else "",
)
self._store.open_work_item(
item_id,
title=title,
scope=GENERAL,
position=READY,
data={"loop": loop.to_data(), "roles": self._roles},
cause="the operator started a conversation",
)
log = self._store.transcript(item_id, "orchestrator")
if before is not None:
closing = [
e.event.text
for e in self._store.transcript(before.id, "orchestrator").read()
if isinstance(e.event, Fact) and e.event.text.startswith("work item ")
]
log.append(
Fact(
text=f'this continues the conversation "{before.title}" '
f"({before.id}), "
+ (closing[-1] if closing else "which is still open")
)
)
log.as_actor("operator").append(
UserMessage(text=text, source=f"transport:{msg.sender}")
)
def _go(self, item: WorkItem, loop: Loop, by: str) -> None:
self._fact(item, f"go-ahead on rev-{loop.rev} from {by}")
if item.scope == GENERAL and loop.proposal_kind == "project":
self._checklist(item, loop)
return
if item.scope == GENERAL:
self._file_work(item, loop)
return
save(
self._store,
item.id,
READY,
replace(loop, started=True, wait="", grace_until=0.0),
cause=f"go-ahead on rev-{loop.rev}",
)
def _file_work(self, item: WorkItem, loop: Loop) -> None:
"""D1: agreed work from a conversation gets a record issue in its
project, carrying the new item's ID from the moment it exists, and
an item of its own that starts as agreed. Under an intent (L8): a
crash between the issue and the item is settled at boot."""
project = self._projects.get(loop.proposal_project)
summary = self._latest_proposal(item)
if project is None or not summary:
note = "that proposal can't be filed any more; what next?"
self._wait(item, replace(loop, wait=ANSWER, note=note), note)
return
new_id = f"{project.name}-c{uuid.uuid4().hex[:6]}"
body = (
f"{summary}\n\nFrom a conversation with the operator, agreed there as "
f"rev-{loop.rev}.\n\n<!-- nisse-item: {new_id} -->"
)
intent = self._store.begin_intent(
FILE_ISSUE,
{"repo": project.repo, "item": new_id, "title": loop.proposal_title},
cause=f"filing {loop.proposal_title}",
work_item=item.id,
)
try:
number = self._gitea.create_issue(project.repo, loop.proposal_title, body)
self._gitea.assign_issue(project.repo, number, self._account)
except GiteaError as exc:
self._store.abort_intent(intent, cause=f"Gitea refused: {exc}")
note = f"I couldn't open the issue: {exc}. Say go to try again."
self._wait(item, replace(loop, wait=GO_AHEAD, note=note), note)
return
self._start_filed(item, project, new_id, number, loop.proposal_title, summary)
self._store.commit_intent(intent, cause=f"opened {project.repo}#{number}")
def _start_filed(
self,
conversation: WorkItem,
project: Project,
new_id: str,
number: int,
title: str,
summary: str,
) -> None:
"""The filed issue's item, agreed already, and the conversation told
where the work went."""
root = self._transport.post(
f"Queued {project.repo}#{number}: {title} (from a conversation)"
)
new = Loop(
thread=root,
issue=number,
labels=project.labels,
started=True, # agreed in the conversation
follows=conversation.id,
)
self._threads[(project.name, number)] = root
self._store.open_work_item(
new_id,
title=title,
scope=project.name,
position=READY,
data={"loop": new.to_data(), "roles": self._roles},
cause=f"filed {project.repo}#{number} from {conversation.id}",
)
self._store.transcript(new_id, "orchestrator").append(
Fact(
text=f"work item opened from {project.repo}#{number} ({title}), "
f"filed from the conversation {conversation.id}, where the operator "
f"agreed to this brief:\n{summary}"
)
)
loop = loop_of(self._store.work_item(conversation.id) or conversation)
loop = replace(
loop,
became=(*loop.became, f"{project.repo}#{number}"),
wait=CHAT,
note="",
proposal_project="",
proposal_title="",
idle_since=self._clock(),
)
self._fact(conversation, f"filed {project.repo}#{number} as {new_id}")
self._wait(
conversation,
loop,
f"Opened {project.repo}#{number} and started on it in its own thread. "
"We can keep talking here.",
)
# -- new projects (D1, D8) --------------------------------------------------
def _checklist(self, item: WorkItem, loop: Loop) -> None:
"""An agreed new project: what it needs, for the operator to provide
(later nisse-ops, O6), with its project entry to add."""
name = loop.proposal_project
entry = (
"projects:\n"
f" {name}:\n"
f" repo: <owner>/{name}\n"
" base_branch: main\n"
f" workspace: /root/work/{name}\n"
" build_host:\n"
" address: <the build VM>\n"
" key: <my SSH key for it>\n"
" push_key: <the deploy key's private half>"
)
self._fact(item, f"provisioning checklist posted for {name}")
self._wait(
item,
replace(
loop,
wait=PROVISION,
note=f"{name}'s setup: say done when it's ready",
new_project=name,
idle_since=self._clock(),
),
f"To set up `{name}`, I need:\n"
f"1. A repository for it, e.g. <owner>/{name}, with a README so its "
"default branch exists.\n"
"2. Protection on the default branch: none of my pushes, and an "
"approval to merge.\n"
f"3. Me ({self._account}) as a collaborator with write access.\n"
"4. A build VM, with the toolchain from the decisions above.\n"
"5. A deploy key with write access, its private half where I run.\n"
"6. This entry in my configuration, with the <...> parts filled "
f"in:\n\n```yaml\n{entry}\n```\n\n"
"Say done when it's ready, and I'll check each of them.",
)
def _provisioned(self, item: WorkItem, loop: Loop) -> None:
""" "done": the entry is read again and the new project checked
(D4: verified, not assumed). What fails is said, per item; when all
is well the tracking issue is opened, and the project's item starts
with its architecture."""
name = loop.new_project
try:
added = list(self._reload(set(self._projects))) if self._reload else []
except Exception as exc: # noqa: BLE001 -- said; the operator fixes it
self._transport.post(
f"I couldn't read the new entry: {exc}. Fix it and say done again.",
thread=loop.thread,
)
return
project = next((p for p in added if p.name == name), None)
if project is None:
self._transport.post(
f"I don't see `{name}` in my configuration yet (item 6). Add its "
"entry and say done again.",
thread=loop.thread,
)
return
failures = (
project.milestones.verify_access(self._account)
if project.milestones is not None
else ["the project has no workspace"]
)
if failures:
self._fact(item, f"{name} checked: " + "; ".join(failures))
self._transport.post(
"Not ready yet: " + "; ".join(failures) + ". Say done again when "
"they're fixed.",
thread=loop.thread,
)
return
self._projects[name] = project
self._fact(item, f"{name} set up and checked: {project.repo}")
brief = self._latest_proposal(item)
new_id = f"{name}-c{uuid.uuid4().hex[:6]}"
title = f"Implement the build plan for {name}"
try:
number = self._gitea.create_issue(
project.repo,
title,
f"{brief}\n\nFrom a conversation with the operator, agreed there "
f"as rev-{loop.rev}. First: ARCHITECTURE.md and the build plan."
f"\n\n<!-- nisse-item: {new_id} -->",
)
self._gitea.assign_issue(project.repo, number, self._account)
except GiteaError as exc:
self._transport.post(
f"{name} checks out, but I couldn't open its tracking issue: "
f"{exc}. Say done to try again.",
thread=loop.thread,
)
return
# D1: the architecture is agreed in its PR, as `design` steers.
design = replace(
project, labels=tuple(dict.fromkeys(("design", *project.labels)))
)
self._start_filed(item, design, new_id, number, title, brief)
loop = loop_of(self._store.work_item(item.id) or item)
save(
self._store,
item.id,
WAITING,
replace(loop, new_project="", proposal_kind=""),
cause=f"{name} provisioned",
)
def _latest_proposal(self, item: WorkItem) -> str:
proposals = [
e.event.plan
for e in self._store.transcript(item.id, "driver").read()
if isinstance(e.event, PlanSubmitted)
]
return proposals[-1] if proposals else ""
def _plain_go(self, item: WorkItem, text: str) -> bool:
"""Whether a reply that isn't a command agrees to the restatement up.
Questions never do (L5)."""
if self._reader is None or is_question(text):
return False
proposals = [
e.event.plan
for e in self._store.transcript(item.id, "driver").read()
if isinstance(e.event, PlanSubmitted)
]
return bool(proposals) and self._reader.agrees(proposals[-1], text)
def _which(self, command: str) -> str:
"""L5: an unnamed command at top level is ambiguous: ask."""
items = [i for i in self._store.work_items("open") if i.scope != GENERAL]
if not items:
return f"{command} what? I have no open work items."
listed = ", ".join(f"{self._label(i)} ({i.title})" for i in items)
return (
f"Which item? Say it in the item's thread, or name it "
f'("{command} {self._label(items[0])}"). Open: {listed}.'
)
def _not_one(
self, found: list[WorkItem], project: str | None, issue: int, say: str
) -> str:
"""A named item that is none, or more than one (the same number in
two projects). ``say`` is the command, with {} for the item."""
name = f"{project}#{issue}" if project else f"#{issue}"
if not found:
return f"{name} isn't one of my open work items."
options = " or ".join(say.format(self._label(i)) for i in found)
return f"{name} is open in more than one project: say {options}."
def _put_first(self, project: str | None, issue: int) -> str:
"""The operator's "do #N next" (L1): it runs before every other
ready item, the latest such order first. It doesn't unblock."""
found = self._named(project, issue)
if len(found) != 1:
return self._not_one(found, project, issue, "do {} next")
[item] = found
name = self._label(item)
items = self._store.work_items("open")
first = max(loop_of(i).first for i in items) + 1
loop = replace(loop_of(item), first=first)
save(self._store, item.id, item.position, loop, cause=f"{name} put first")
self._fact(item, "the operator put this item first in the queue")
if item.position == BLOCKED:
return (
f"{name} goes first once it's unblocked "
f"(blocked by {', '.join(loop.blocked_by)})."
)
if item.position == WAITING:
return f"{name} goes first, but it's waiting for {loop.wait}: {loop.note}"
return f"{name} is next."
def _queue(self) -> list[WorkItem]:
"""Ready items in the order they run (L1): the operator's order
first, then the oldest."""
items = self._store.work_items("open") # oldest first
ready = [i for i in items if i.position == READY]
return sorted(ready, key=lambda i: -loop_of(i).first)
def status(self) -> str:
items = self._store.work_items("open")
if not items:
return (
"idle: no open work items. I take issues assigned to me in "
f"{self._repos()}."
)
queue = [i.id for i in self._queue()]
lines = []
for item in items:
loop = loop_of(item)
what = item.position
if item.position == WAITING and loop.wait == CHAT:
what = "talking; your turn"
elif item.position == WAITING:
what = f"waiting for {loop.wait} ({loop.note or f'rev-{loop.rev}'})"
if loop.grace_until:
left = max(0, int(loop.grace_until - self._clock()) // 60)
what += f", going ahead in {left} min unless you reply"
elif item.position == BLOCKED:
what = f"blocked by {', '.join(loop.blocked_by)}"
elif item.position == READY:
what = f"ready, {queue.index(item.id) + 1} of {len(queue)} in line"
name = (
f'conversation "{item.title}"'
if item.scope == GENERAL
else f"{self._label(item)} {item.title}"
)
lines.append(f"{name}: {what}")
return "\n".join(lines)
# -- running (L1, L2) ------------------------------------------------------
def _schedule(self) -> None:
"""Run ready items until none is left. Only on the worker."""
if self._running:
return
while True:
ready = self._queue()
if not ready:
self._store.set_active(None, cause="nothing ready")
return
self._run(ready[0])
def _keep(self, project: str) -> tuple[str, ...]:
"""Branch prefixes of a project's open items: parked or not, they
are never set aside."""
return tuple(
f"{i.id}-" for i in self._store.work_items("open") if i.scope == project
)
def _park_elsewhere(self, here: str) -> None:
"""L1: switching to another project's item parks the one on the
project worked on last, so nothing lives only on its build VM. After
a boot nobody knows which that was, so every other is parked."""
last = self._last_project
others = [p for p in self._projects.values() if p.name != here]
for other in others if last is None else [self._projects[last]]:
if other.name == here or other.milestones is None:
continue
try:
other.milestones.park(keep=self._keep(other.name))
except Exception as exc: # noqa: BLE001 -- reported; the run goes on
self._oplog.emit(
"loop.park_failed", cause=repr(exc), project=other.name
)
self._last_project = here
def _converse_run(self, item: WorkItem) -> None:
"""A conversation's turn (L1, L5): it talks, and may propose work on
a project. No workspace, nothing to run."""
loop = loop_of(item)
self._store.set_active(item.id, cause="ready")
save(self._store, item.id, RUNNING, loop, cause="driver run")
projects = "\n".join(f"- {p.name}: {p.repo}" for p in self._projects.values())
system = CONVERSATION.format(projects=f"The projects:\n{projects}")
self._active_thread = loop.thread
self._running = True
try:
outcome = self._driver.run(
self._store.transcript(item.id, "driver"),
system=system,
tools=[REPLY, ASK_HUMAN, PROPOSE_WORK, PROPOSE_PROJECT],
workspace=None,
checks={
"propose_work": self._proposal_problem,
"propose_project": self._project_problem,
},
)
finally:
self._running = False
self._cancel.clear()
self._settle(item.id, outcome)
def _project_problem(self, args: dict[str, object]) -> str | None:
name = str(args.get("name", "")).strip()
if not re.fullmatch(r"[a-z0-9][a-z0-9-]*", name) or name == GENERAL:
return "a project's name is lowercase letters, digits and -, not general"
if name in self._projects:
return f"{name} is a project already: propose_work for work on it"
if not str(args.get("purpose", "")).strip():
return "say what the project is for"
return _restatement_problem(
{"text": f"{args.get('purpose', '')}\n{args.get('decisions', '')}"}
)
def _proposal_problem(self, args: dict[str, object]) -> str | None:
project = str(args.get("project", "")).strip()
if project not in self._projects:
names = ", ".join(self._projects)
return f"no project {project!r}: the projects are {names}"
if not str(args.get("title", "")).strip():
return "the issue needs a title"
summary = str(args.get("summary", "")).strip()
if not summary:
return "the issue needs a summary: what was decided"
return _restatement_problem({"text": summary})
def _run(self, item: WorkItem) -> None:
if item.scope == GENERAL:
self._converse_run(item)
return
loop = loop_of(item)
project = self._at(item)
milestones = project.milestones
self._store.set_active(item.id, cause="ready")
self._park_elsewhere(project.name)
if milestones is not None:
keep = self._keep(project.name)
try:
if not loop.prepared:
# D3: a fresh workspace when a work item starts.
notes = milestones.prepare(f"{item.id}-work", keep=keep)
else:
# D4: back to its branch, parking whoever was there.
notes = []
milestones.resume(
loop.branch or f"{item.id}-work", keep=keep, mine=f"{item.id}-"
)
except Exception as exc: # noqa: BLE001 -- the item waits, not the loop
note = f"I couldn't prepare the workspace: {exc}. Reply to retry."
self._wait(item, replace(loop, wait=ANSWER, note=note), note)
self._store.set_active(None, cause="workspace failed")
return
if not loop.prepared:
self._fact(item, "workspace prepared from the latest default branch")
if notes:
said = "Set aside from the workspace: " + "; ".join(notes)
self._fact(item, said)
self._transport.post(said, thread=loop.thread)
loop = replace(loop, prepared=True)
save(self._store, item.id, RUNNING, loop, cause="driver run")
# D1: a `discuss` item talks before it restates.
hold = "discuss" in loop.labels and not loop.talked and not loop.started
tools = [
*([] if hold else [RESTATE]),
ASK_HUMAN,
*(
_LOOK
if not loop.started
else [
START_MILESTONE,
FINISH_MILESTONE,
CLOSE_ITEM,
*([WAIT_FOR_REVIEW] if loop.pr else []),
*EXECUTOR_TOOLS,
]
),
]
workspace = project.workspace
system = SYSTEM.format(
repo=project.repo,
issue=loop.issue,
title=item.title,
labels="".join(f"{LABELS[x]}\n\n" for x in loop.labels if x in LABELS),
workspace=workspace.describe() if workspace else "",
)
self._active_thread = loop.thread
self._running = True
try:
outcome = self._driver.run(
self._store.transcript(item.id, "driver"),
system=system,
tools=tools,
workspace=workspace,
handlers={"start_milestone": lambda args: self._start(item.id, args)},
checks={"restate": _restatement_problem},
)
finally:
self._running = False
self._cancel.clear()
self._settle(item.id, outcome)
def _settle(self, item_id: str, outcome: RunOutcome) -> None:
item = self._store.work_item(item_id)
assert item is not None
loop = loop_of(item)
args = outcome.arguments or {}
if outcome.kind == "wait" and outcome.tool == "restate":
text = str(args.get("text", "")).strip()
rev = loop.rev + 1
self._store.transcript(item.id, "driver").append(
PlanSubmitted(plan=text, model=self._roles.get("planner", ""))
)
self._fact(item, f"restatement rev-{rev} put up for a go-ahead")
self._shown[loop.thread] = rev
grace = self._clock() + self._grace if "bug" in loop.labels else 0.0
after = "Say go to start, or reply with changes."
if grace:
after = (
f"It's a bug, so I'll go ahead in {round(self._grace / 60)} min "
"unless you reply; say go to start now, or reply with changes."
)
self._transport.post(f"rev-{rev}: {text}\n\n{after}", thread=loop.thread)
self._wait(
item,
replace(loop, rev=rev, wait=GO_AHEAD, note="", grace_until=grace),
None,
)
elif outcome.kind == "wait" and outcome.tool == "reply":
text = str(args.get("text", "")).strip()
self._transport.post(text, thread=loop.thread)
self._wait(
item,
replace(loop, wait=CHAT, note="", idle_since=self._clock()),
None,
)
elif outcome.kind == "wait" and outcome.tool == "propose_work":
loop = replace(loop, proposal_kind="work")
project = self._projects[str(args.get("project", "")).strip()]
title = str(args.get("title", "")).strip()
summary = str(args.get("summary", "")).strip()
rev = loop.rev + 1
self._store.transcript(item.id, "driver").append(
PlanSubmitted(plan=summary, model=self._roles.get("planner", ""))
)
self._fact(item, f"work proposed as rev-{rev}: {project.repo}, {title}")
self._shown[loop.thread] = rev
self._transport.post(
f"rev-{rev}: I'd open an issue in {project.repo}, "
f'"{title}":\n\n{summary}\n\nSay go to open it and start, or '
"reply with changes.",
thread=loop.thread,
)
self._wait(
item,
replace(
loop,
rev=rev,
wait=GO_AHEAD,
note="",
proposal_project=project.name,
proposal_title=title,
idle_since=self._clock(),
),
None,
)
elif outcome.kind == "wait" and outcome.tool == "propose_project":
name = str(args.get("name", "")).strip()
purpose = str(args.get("purpose", "")).strip()
decisions = str(args.get("decisions", "")).strip()
brief = f"{purpose}\n\nDecisions:\n{decisions}" if decisions else purpose
rev = loop.rev + 1
self._store.transcript(item.id, "driver").append(
PlanSubmitted(plan=brief, model=self._roles.get("planner", ""))
)
self._fact(item, f"new project proposed as rev-{rev}: {name}")
self._shown[loop.thread] = rev
self._transport.post(
f"rev-{rev}: a new project, `{name}`. {brief}\n\nSay go and I'll "
"list what it needs set up, or reply with changes.",
thread=loop.thread,
)
self._wait(
item,
replace(
loop,
rev=rev,
wait=GO_AHEAD,
note="",
proposal_project=name,
proposal_title="",
proposal_kind="project",
idle_since=self._clock(),
),
None,
)
elif outcome.kind == "wait" and outcome.tool == "ask_human":
question = str(args.get("question", "")).strip()
self._answer_where_asked(item, loop, question)
self._wait(item, replace(loop, wait=ANSWER, note=question), None)
elif outcome.kind == "wait" and outcome.tool == "wait_for_review":
note = str(args.get("note", "")).strip() or f"waiting on PR #{loop.pr}"
if loop.pr:
self._answer_where_asked(item, loop, note)
self._wait(item, replace(loop, wait=MERGE, note=f"PR #{loop.pr}"), None)
else:
say = f"{note} (there's no open PR, so I need you to tell me what next)"
self._wait(item, replace(loop, wait=ANSWER, note=say), say)
elif outcome.kind == "wait" and outcome.tool == "finish_milestone":
self._finish(
item,
loop,
str(args.get("summary", "")).strip(),
str(args.get("title", "")).strip(),
# Unsaid, it's final: the PR closes the issue, as before.
str(args.get("final", "yes")).strip().lower() != "no",
)
elif outcome.kind == "wait" and outcome.tool == "close_item" and loop.pr:
# Closing leaves nothing open (L2): an open PR is waited out first.
self._fact(
item,
f"close_item waits: PR #{loop.pr} is still open; the item closes "
"after it merges",
)
self._wait(
item,
replace(loop, wait=MERGE, note=f"PR #{loop.pr}"),
f"I'll close this once PR #{loop.pr} merges: {loop.pr_url}",
)
elif outcome.kind == "wait" and outcome.tool == "close_item":
summary = str(args.get("summary", "")).strip()
self._transport.post(f"Done: {summary}", thread=loop.thread)
self._close(item, loop, "completed", summary)
elif outcome.kind == "stopped":
note = f"stopped ({outcome.note}). Why did you stop it? Reply to continue."
self._wait(item, replace(loop, wait=ANSWER, note=note), note)
else:
note = f"I stopped short: {outcome.note}. Reply to continue."
self._wait(item, replace(loop, wait=ANSWER, note=note), note)
self._store.set_active(None, cause="item waiting")
# -- milestones (D4) -----------------------------------------------------
def _start(self, item_id: str, args: dict[str, object]) -> tuple[str, bool]:
"""start_milestone, mid-run: cut the milestone's branch."""
item = self._store.work_item(item_id)
assert item is not None
loop = loop_of(item)
project = self._at(item)
if loop.pr:
# D4: the next milestone waits for the merge, read from Gitea now,
# not from the last watch round.
try:
pr = self._gitea.pull_request(project.repo, loop.pr)
except GiteaError as exc:
return f"couldn't check PR #{loop.pr}: {exc}; try again", False
if pr.state == "open":
wait = (
f"milestone {loop.milestone}'s PR #{loop.pr} hasn't merged yet; "
"the next milestone starts after it merges. If you're only "
"waiting for that, call wait_for_review."
)
return wait, False
how = "merged" if pr.merged else "was closed without merging"
self._fact(item, f"PR #{loop.pr} {how}")
loop = replace(loop, pr=0, pr_url="", branch="")
if loop.branch:
busy = f"milestone {loop.milestone} is still open; finish it first"
return busy, False
if project.milestones is None:
return "milestones aren't available here", False
number = loop.milestone + 1
branch = f"{item.id}-milestone-{number}-{uuid.uuid4().hex[:4]}"
try:
project.milestones.start(branch)
except GitError as exc:
return f"couldn't start the milestone: {exc}", False
name = str(args.get("name", "")).strip()
self._fact(item, f"milestone {number} started ({name}) on branch {branch}")
save(
self._store,
item.id,
RUNNING,
replace(loop, milestone=number, branch=branch),
cause=f"milestone {number} started",
)
return f"started milestone {number} on branch {branch}", True
def _answer_where_asked(self, item: WorkItem, loop: Loop, text: str) -> None:
"""D5: answer where the operator asked. A PR question is answered on
the PR, quoting the line it was about (an inline thread can't be
replied to through the API), with a pointer in the Matrix thread."""
if not loop.asked_on_pr:
self._transport.post(text, thread=loop.thread)
return
quoted = [f"> on `{loop.asked_where}`:"] if loop.asked_where else []
quoted += [f"> {line}" for line in loop.asked_text.splitlines()]
body = "\n".join([*quoted, "", text])
try:
self._gitea.post_comment(self._at(item).repo, loop.asked_on_pr, body)
except GiteaError:
self._transport.post(text, thread=loop.thread) # the PR is unreachable
return
url = loop.pr_url or f"PR #{loop.asked_on_pr}"
self._transport.post(
f"I answered your comment on PR #{loop.asked_on_pr}: {url}",
thread=loop.thread,
)
def _finish(
self,
item: WorkItem,
loop: Loop,
summary: str,
title: str = "",
final: bool = True,
) -> None:
"""finish_milestone: publish the milestone, or push a review fix."""
project = self._at(item)
milestones = project.milestones
if milestones is None:
note = "milestones aren't available here"
self._wait(item, replace(loop, wait=ANSWER, note=note), note)
return
if loop.pr:
fixed = milestones.push_fix(
item_id=item.id,
branch=loop.branch,
pr=loop.pr,
message=summary or "Address review comments",
attempt=loop.fixes + 1,
)
self._fact(item, f"review fix: {fixed.note}")
if fixed.kind in ("pushed", "nothing"):
self._gitea_comment(project.repo, loop.pr, fixed.note)
self._wait(
item,
replace(loop, wait=MERGE, note=f"PR #{loop.pr}"),
f"PR #{loop.pr}: {fixed.note}",
)
elif fixed.kind == "new-pr" and fixed.pr is not None:
# Later fixes go to the new PR's branch, never the merged one.
loop = replace(
loop,
pr=fixed.pr.number,
pr_url=fixed.pr.url,
branch=fixed.branch,
fixes=loop.fixes + 1,
)
self._wait(
item,
replace(loop, wait=MERGE, note=f"PR #{fixed.pr.number}"),
f"The PR merged before the fix landed, so it's a new PR: "
f"#{fixed.pr.number} {fixed.pr.url}",
)
else:
self._wait(
item,
replace(loop, wait=ANSWER, note=fixed.note),
f"{fixed.note}. What next?",
)
return
if not loop.branch:
# D4: finishing without a start cuts the branch now.
number = loop.milestone + 1
branch = f"{item.id}-milestone-{number}-{uuid.uuid4().hex[:4]}"
try:
milestones.start(branch)
except GitError as exc:
note = f"I couldn't start milestone {number}: {exc}"
self._wait(item, replace(loop, wait=ANSWER, note=note), note)
return
loop = replace(loop, milestone=number, branch=branch)
published = milestones.publish(
item_id=item.id,
branch=loop.branch,
title=title or f"{item.title} (milestone {loop.milestone})",
summary=summary,
issue=loop.issue,
final=final,
)
left = (
f" (left out {len(published.left_out)} build artifact(s))"
if published.left_out
else ""
)
if published.pr is not None and published.kind in ("opened", "adopted"):
pr = published.pr
self._fact(
item,
f"milestone {loop.milestone} published: PR #{pr.number} "
f"({pr.url}){left}",
)
self._wait(
item,
replace(
loop,
pr=pr.number,
pr_url=pr.url,
wait=MERGE,
note=f"PR #{pr.number}",
),
f"Milestone {loop.milestone} is up for review: {pr.url}{left}",
)
else:
self._fact(
item, f"milestone {loop.milestone} not published: {published.note}"
)
self._wait(
item,
replace(loop, wait=ANSWER, note=published.note),
f"Milestone {loop.milestone}: {published.note}. What next?",
)
def _gitea_comment(self, repo: str, pr: int, text: str) -> None:
try:
self._gitea.post_comment(repo, pr, text)
except GiteaError:
pass # courtesy only; the pushed commit is the answer
def _wait(self, item: WorkItem, loop: Loop, say: str | None) -> None:
if say:
self._transport.post(say, thread=loop.thread)
save(self._store, item.id, WAITING, loop, cause=f"waiting for {loop.wait}")
def _close(self, item: WorkItem, loop: Loop, outcome: str, note: str) -> None:
self._fact(item, f"work item {outcome}: {note}")
project = self._at(item)
if loop.pr and outcome == "abandoned":
# L2: abandoning closes the item's open PR with a note.
try:
self._gitea.post_comment(project.repo, loop.pr, f"abandoned: {note}")
self._gitea.close_pull_request(project.repo, loop.pr)
except GiteaError as exc:
self._transport.post(
f"could not close PR #{loop.pr}: {exc}", thread=loop.thread
)
save(self._store, item.id, CLOSED, loop, cause=outcome)
self._store.close_work_item(item.id, outcome=outcome, cause=note)
if project.milestones is not None:
try:
project.milestones.drop_parked(f"{item.id}-")
except GitError:
pass # a leftover side branch; nothing depends on it
if loop.issue:
try:
self._gitea.post_comment(project.repo, loop.issue, f"{outcome}: {note}")
self._gitea.close_issue(project.repo, loop.issue)
except GiteaError as exc:
self._transport.post(
f"closed here, but could not close #{loop.issue}: {exc}",
thread=loop.thread,
)
if outcome == "abandoned":
self._transport.post(f"abandoned: {note}", thread=loop.thread)
def _fact(self, item: WorkItem, text: str) -> None:
self._store.transcript(item.id, "orchestrator").append(Fact(text=text))
Reference in new issue View Git Blame Copy Permalink