Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
101 changes: 86 additions & 15 deletions bin/fm_open_loops.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
r"|ssh://(?:git@)?github[.]com/|git@github[.]com:)"
r"([^/]+/[^/]+?)(?:[.]git)?")
PIPELINE_ENDED = {"completed", "failed", "cancelled", "aborted"}
STOP_STATES = {"done", "parked", "paused"}
SIGNATURE = re.compile(r"usage.?limit|rate.?limit|quota|auth|unauthorized|login|network|connection"
r"|timed? ?out|ECONN|ENOTFOUND|\b(?:error|failure|failed|exception|fatal)\b"
r"|\b(?-i:E[A-Z][A-Z0-9_]{2,})\b", re.I)
Expand Down Expand Up @@ -83,6 +84,21 @@ def __init__(self, home, now):
self.backlog = []
self.prs = {}
self.state_files = []
self.owned_urls = set()
self.owned_branches = set()
self.lane = self.is_secondmate_home()

def is_secondmate_home(self):
"""A genuine secondmate-home marker: a regular file holding a valid id (fm-primary-scope-lib.sh)."""
marker = self.home / ".fm-secondmate-home"
try:
if marker.is_symlink() or not marker.is_file():
return False
first = marker.read_text().splitlines()[:1]
except (OSError, UnicodeDecodeError):
return False
ident = re.sub(r"\s", "", first[0]) if first else ""
return bool(ident) and re.fullmatch(r"[A-Za-z0-9._-]+", ident) is not None

def load_config(self):
path = self.config / "open-loops.json"
Expand Down Expand Up @@ -194,13 +210,17 @@ def git(self, repo, *args):
def bash(self, script, *args):
return self.run(["bash", "-c", script, "open-loops", *args])

def add(self, category, subject, action, since=None, owner="firstmate", evidence=""):
def add(self, category, subject, action, since=None, owner="firstmate", evidence="", informational=False):
"""An informational row is visible context, never an obligation: it is not overdue at any age."""
identity = hashlib.sha256(f"{self.home}\0{category}\0{subject}".encode()).hexdigest()[:24]
age = None if since is None or since > self.now else self.now - since
limit = self.ages[category]
self.rows[identity] = dict(id=identity, category=category, subject=str(subject), owner=owner,
next_action=action, age_seconds=age, limit_seconds=limit,
overdue=age is None or age >= limit, evidence=str(evidence)[:400])
row = dict(id=identity, category=category, subject=str(subject), owner=owner,
next_action=action, age_seconds=age, limit_seconds=limit,
overdue=not informational and (age is None or age >= limit), evidence=str(evidence)[:400])
if informational:
row["informational"] = True
self.rows[identity] = row

def source(self, name, reader, *args):
"""Run one reader; its failure marks the ledger partly blind and never aborts the others."""
Expand All @@ -227,20 +247,42 @@ def read_snapshot(self):
if not record.get("structured") and record.get("state") != "done":
self.degraded.append("unstructured backlog row: " + str(record.get("raw", ""))[:100])

def recorded_stop(self, task):
"""True when the worker's own durable record says it stopped on purpose, so its endpoint
being gone or its live state being unreadable is expected rather than a loss."""
if task["current_state"].get("detail") == "task generation changed during snapshot":
return False
state = task["current_state"].get("state")
if state in ("working", "blocked", "failed"):
return False
if state in STOP_STATES:
return True
return self.bash('. "$1"; last=$(last_status_line "$2"); '
'if [ "$(status_line_verb "$last")" = done ] || '
'status_is_paused "$(status_declared_wait_line "$2")"; then printf stopped; fi',
BIN / "fm-status-event-lib.sh", self.state / (task["id"] + ".status")).strip() == "stopped"

def backlog_rows(self):
by_id = {t["id"]: t for t in self.tasks}
dispatched = {p.name.rsplit(".", 1)[0] for p in self.state_files if p.name.endswith((".meta", ".status"))}
today = dt.datetime.fromtimestamp(self.now, dt.timezone.utc).strftime("%Y-%m-%d")
for record in self.backlog:
since = epoch(record.get("since"))
task = by_id.get(record["id"])
if record["state"] == "in_flight" and record.get("requires_child_metadata"):
if not task or task["endpoint"].get("exists") is False:
if not task or (task["endpoint"].get("exists") is False and not self.recorded_stop(task)):
self.add("missing_worker", record["id"], "recover the assigned worker without discarding work",
since)
elif (record["state"] == "queued" and not record.get("hold_kind")
and not record.get("unresolved_blocker_ids")
and (not record.get("hold_until") or record["hold_until"] <= today)):
self.add("ready_not_started", record["id"], "dispatch the dependency-cleared work", since)
# Only work whose dispatch began (a task record or status log exists) is an obligation;
# the rest of the queue is the captain's backlog, shown but never overdue.
selected = record["id"] in by_id or record["id"] in dispatched
self.add("ready_not_started", record["id"],
"dispatch the dependency-cleared work" if selected
else "informational: queued and not selected for dispatch",
since, informational=not selected)

def liveness(self, task):
alive = task["endpoint"].get("agent_alive")
Expand All @@ -257,13 +299,20 @@ def liveness(self, task):

def worker_rows(self):
for task in self.tasks:
alive = self.source("worker liveness " + task["id"], self.liveness, task)
if (task["current_state"].get("state") == "unknown"
stopped = self.source("worker stop " + task["id"], self.recorded_stop, task)
if stopped: # a recorded stop explains an unreadable endpoint; it never degrades coverage
try:
alive = self.liveness(task)
except SOURCE_ERRORS:
alive = None
else:
alive = self.source("worker liveness " + task["id"], self.liveness, task)
if (task["current_state"].get("state") == "unknown" and not stopped
and not (alive in ("dead", "missing")
and task["current_state"].get("detail", "").startswith("backend target gone:"))):
self.degraded.append("worker current state " + task["id"] + ": "
+ (task["current_state"].get("detail") or "observation unavailable"))
if task["endpoint"].get("exists") is False or alive in ("dead", "missing"):
if not stopped and (task["endpoint"].get("exists") is False or alive in ("dead", "missing")):
self.add("missing_worker", task["id"], "recover the assigned worker without discarding work",
mtime(self.state / (task["id"] + ".status")) or mtime(self.state / (task["id"] + ".meta")))
if task["current_state"].get("state") == "failed":
Expand Down Expand Up @@ -442,14 +491,32 @@ def pr_state(self, url):
self.prs[url] = pr
return pr

def load_ownership(self):
"""PRs this home owns: recorded on its tasks or backlog, or opened from a task's branch."""
self.owned_urls = {t["pr"]["url"] for t in self.tasks if t["pr"].get("url")}
for record in self.backlog:
for link in [record.get("pr_url"), *(record.get("links") or [])]:
if isinstance(link, str) and GITHUB_PR.fullmatch(link):
self.owned_urls.add(link)

def owns_pr(self, pr):
head = pr.get("head") or {}
return pr["html_url"] in self.owned_urls or \
(((head.get("repo") or {}).get("full_name") or "").casefold(), head.get("ref")) in self.owned_branches

def repo_slugs(self):
paths = {t["project"] for t in self.tasks if t.get("project")}
if (self.home / ".git").exists():
paths.add(str(self.home))
if self.projects_dir.is_dir():
projects = self.source("projects inventory", lambda: list(self.projects_dir.iterdir())) or []
paths |= {str(p) for p in projects if (p / ".git").exists()}
if not self.lane: # a lane's shared clones are not its obligations; only the main home surveys them
if (self.home / ".git").exists():
paths.add(str(self.home))
if self.projects_dir.is_dir():
projects = self.source("projects inventory", lambda: list(self.projects_dir.iterdir())) or []
paths |= {str(p) for p in projects if (p / ".git").exists()}
slugs = set()
for url in self.owned_urls:
match = GITHUB_PR.fullmatch(url)
if match:
slugs.add(match[1])
for task in self.tasks:
url = task["pr"].get("url")
if not url:
Expand All @@ -466,6 +533,8 @@ def repo_slugs(self):
match = GITHUB_ORIGIN.fullmatch(origin)
if match:
slugs.add(match[1])
self.owned_branches.update((match[1].casefold(), t["branch"]) for t in self.tasks
if t.get("project") == path and t.get("branch"))
return sorted(slugs)

def repo_origin(self, path):
Expand Down Expand Up @@ -523,7 +592,8 @@ def pr_rows(self, slug):
self.pr_state(url)
continue
self.prs[url] = pr
self.source(f"PR {slug}#{pr.get('number', '?')}", self.pr_row, slug, pr)
if self.owns_pr(pr):
self.source(f"PR {slug}#{pr.get('number', '?')}", self.pr_row, slug, pr)

def question_rows(self):
for record in self.backlog:
Expand Down Expand Up @@ -553,6 +623,7 @@ def collect(self):
self.source("state inventory", self.read_state_inventory)
self.source("fleet snapshot", self.read_snapshot)
self.source("backlog", self.backlog_rows)
self.source("ownership", self.load_ownership)
slugs = self.source("project origins", self.repo_slugs) or []
for slug in slugs:
self.source("open PRs " + slug, self.pr_rows, slug)
Expand Down
3 changes: 2 additions & 1 deletion docs/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -472,7 +472,8 @@ For manual project initialization, [`bin/fm-ensure-agents-md.sh`](../bin/fm-ensu
The destination for each kind of knowledge, including project-intrinsic knowledge, is owned by [`AGENTS.md` section 6](../AGENTS.md#6-project-and-knowledge-management).
Memory writes use inspect-then-update rather than blind append; the internal [`stow` skill](../.agents/skills/stow/SKILL.md) owns tier markers, decay, cold archival, and offload.
The same pass also persists open-work record state the session is holding - filing a thread that was never recorded and correcting one the session knows went stale - bounded to the open work that session is actually holding.
It is deliberately not a reconciliation of durable records against repository or PR reality: its input is the volatile context, so it can only preserve what the session still knows, and no reconciliation that outlives a session exists today.
It is deliberately not a reconciliation of durable records against repository or PR reality: its input is the volatile context, so it can only preserve what the session still knows.
The [open-work ledger](configuration.md#open-work-ledger-configopen-loopsjson) owns reconciliation against live worker and delivery evidence.
Task-scoped notes use `bin/fm-tasks-axi.sh show <id> --full` followed by `bin/fm-tasks-axi.sh update <id> --body-file <path>`, adding `--archive-body` when the prior body should remain recoverable.
The stow pass never writes a skill, but a separately executed, captain-approved migration may move conditional knowledge into a user-owned local skill excluded from the Firstmate clone; changes to Firstmate's tracked skills remain deliberate repository work through the normal PR pipeline.
Invoked in a primary home, `/stow` then cascades the same sweep to every registered secondmate, enumerated through `bin/fm-stow-cascade.sh`: each home is accounted and curated against its own startup-memory allowance, a live secondmate sweeps its own session, and a slow or unreachable home is reported as an exception rather than blocking the primary.
Expand Down
Loading
Loading