Skip to content

feat(claude): absorb grandchild activity into teammate liveness and lead state #352

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gcko merged 1 commit into from
Sep 15, 2026
+107 −36
Merged

feat(claude): absorb grandchild activity into teammate liveness and lead state #352

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
81 changes: 45 additions & 36 deletions cargento/skills/cargento/cargento_runtime/collectors/claude.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -559,6 +559,34 @@ def collect(
# state derivation takes the running ones straight back out, so the two
# lists retain the measured distinction. See
# [unchanged-state capture](docs/captures/README.md#files).
member_flags = {
key: m["active"]
for m in team_members.get(prefix, [])
for key in (m["label"], m["agent_id"], m["local"])
}
child_agent_files = {
c["path"]: agent_transcripts(c["path"], config=config, state=state) for c in children
}
child_is_live: dict[str, bool] = {}
for c in children:
name = c.get("agent_name") or ""
g_files = child_agent_files.get(c["path"], [])
has_live_grandchild = any(
runtime_sessions.is_fresh(config, now, gm, config.working_threshold_sec)
for _, gm in g_files
)
live = (
runtime_sessions.is_fresh(config, now, c["mtime"], config.working_threshold_sec)
or has_live_grandchild
)
# Demote only. A retained-but-finished member is marked inactive and
# that is worth believing; the reverse is not, because the flag on a
# hard-killed pane is unmeasured and DRC-4229 keeps transcript
# freshness authoritative for "running now".
if live and member_flags.get(name) is False:
live = False
child_is_live[c["path"]] = live

own_agents = load_subagents(
config,
transcript,
Expand All @@ -573,14 +601,15 @@ def collect(
*(
{
"label": c["label"],
"mtime": c["mtime"],
"mtime": max(
[c["mtime"], *(gm for _, gm in child_agent_files.get(c["path"], []))],
default=c["mtime"],
),
"model": models.get(c["path"]),
"started_at": child_started[c["path"]],
}
for c in children
if runtime_sessions.is_fresh(
config, now, c["mtime"], config.working_threshold_sec
) # fresh = running
if child_is_live[c["path"]] # fresh = running
),
]
latest_agent_mtime = max(
Expand All @@ -592,12 +621,17 @@ def collect(
# running: a workflow that has been going for hours parks its parent
# transcript, and without this the session ages out of the window.
latest_agent_file_mtime = max((m for _, m in agent_files), default=0)
latest_grandchild_file_mtime = max(
(gm for files in child_agent_files.values() for _, gm in files),
default=0,
)
activity_sources = (
latest_task_mtime,
transcript_mtime,
latest_agent_mtime,
latest_agent_file_mtime,
latest_child_mtime,
latest_grandchild_file_mtime,
)
last_activity = runtime_sessions.newest_plausible(config, now, activity_sources)
active = runtime_sessions.is_fresh(config, now, last_activity, window_hours * 3600)
Expand Down Expand Up @@ -632,18 +666,11 @@ def collect(
# strip, `last_activity` -- keeps reading `subagents`, which is gated on
# `working_threshold_sec` because DRC-4118 settled that a child parked
# hours ago must not make its parent read "running 1 subagent", and
# DRC-4263 requires it. What a person reads on the row is a
# different question: a teammate blocked on its own subagents writes
# nothing for minutes and used to flicker off the row entirely. It now
# stays, and reads as stopped -- as does an agent the lead dispatched
# itself, which review round 1 found still fresh-gated while the
# children and grandchildren beside it were not. That is not
# the whole of what a reader wants. A teammate whose own worker is
# writing is alive, and nothing here says so: the honest fix absorbs a
# grandchild's mtime into this session's activity the way DRC-4118 does
# for Cursor and `latest_agent_file_mtime` already does for the lead's
# own agents, and that moves `state`, which this issue's gate froze.
# Filed rather than taken.
# DRC-4263 requires it. A teammate blocked on its own subagents writes
# nothing for minutes; DRC-4346 absorbs its live grandchild workers'
# activity into the teammate's liveness and this session's activity,
# so the teammate reads as active and the lead reads as working while
# its workers are running.
roster: list[dict[str, Any]] = [
published_agent(
a["label"],
Expand All @@ -654,32 +681,14 @@ def collect(
)
for a in own_agents
]
# Keyed on `label`, which prefers the registry's own `name` -- the same
# string a child transcript writes as `agentName` and the same one
# `started_agent_ids` joins on three calls up. Keying on the `agentId`
# halves instead only agreed while `agentId` was `<name>@session-<prefix>`,
# and a member whose id is a hex handle would have silently escaped the
# demotion.
member_flags = {
key: m["active"]
for m in team_members.get(prefix, [])
for key in (m["label"], m["agent_id"], m["local"])
}
for c in children:
name = c.get("agent_name") or ""
live = runtime_sessions.is_fresh(config, now, c["mtime"], config.working_threshold_sec)
# Demote only. A retained-but-finished member is marked inactive and
# that is worth believing; the reverse is not, because the flag on a
# hard-killed pane is unmeasured and DRC-4229 keeps transcript
# freshness authoritative for "running now".
if live and member_flags.get(name) is False:
live = False
roster.append(
published_agent(
c["label"],
model=models.get(c["path"]),
started_at=child_started[c["path"]],
active=live,
active=child_is_live[c["path"]],
parent=None,
)
)
Expand All @@ -690,7 +699,7 @@ def collect(
# orphan a grandchild into a peer row. They reach the roster only:
# counting them into `subagents` would make a lead running one
# teammate read as running eight.
for gp, gm in agent_transcripts(c["path"], config=config, state=state):
for gp, gm in child_agent_files.get(c["path"], []):
if not runtime_sessions.is_fresh(config, now, gm, window_hours * 3600):
continue
roster.append(
Expand Down
62 changes: 62 additions & 0 deletions cargento/skills/cargento/tests/test_claude.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3296,3 +3296,65 @@ def test_every_published_element_declares_both_new_keys(self) -> None:
self.assertEqual(["ensign-pending"], [a["name"] for a in session["subagents"]])
self.assertIs(False, session["subagents"][0]["active"])
self.assertIsNone(session["subagents"][0]["parent"])

def test_a_teammate_blocked_on_live_workers_reads_as_running_and_keeps_lead_working(
self,
) -> None:
# DRC-4346. A teammate quiet for 300 seconds whose grandchild worker
# is actively writing reads as running, attributes the worker, and keeps
# the lead session working with its last activity absorbed.
now = time.time()
teammate_stamp = datetime.fromtimestamp(now - 300, UTC).isoformat()
lens_stamp = datetime.fromtimestamp(now - 5, UTC).isoformat()
child_sid = "bbbb2222-0000-0000-0000-000000000000"
with tempfile.TemporaryDirectory() as tmp:
proj = self.project(tmp, now=now)
self.teammate(
proj, sid=child_sid, name="ensign-review", stamp=teammate_stamp, age=300, now=now
)
lenses = proj / child_sid / "subagents"
lenses.mkdir(parents=True)
lens_fp = lenses / "agent-lens.jsonl"
lens_fp.write_text(
json.dumps({"type": "user", "timestamp": lens_stamp, "message": {}}) + "\n"
)
(lenses / "agent-lens.meta.json").write_text(json.dumps({"name": "lens-worker"}))
os.utime(lens_fp, (now - 5, now - 5))
session = self.collect_one(tmp, None, now)

published = {a["name"]: a for a in session["subagents"]}
self.assertEqual({"ensign-review", "lens-worker"}, set(published))
self.assertIs(True, published["ensign-review"]["active"])
self.assertIs(True, published["lens-worker"]["active"])
self.assertEqual("ensign-review", published["lens-worker"]["parent"])
self.assertEqual("working", session["state"])
self.assertEqual("running 1 subagent", session["state_detail"])
self.assertEqual(now - 5, session["last_activity"])

def test_a_teammate_with_only_stale_workers_reads_as_not_running(self) -> None:
# DRC-4346. When both teammate and its workers are quiet, the teammate
# still reads as not running, and does not make the lead read as working.
now = time.time()
stale_stamp = datetime.fromtimestamp(now - 300, UTC).isoformat()
child_sid = "bbbb2222-0000-0000-0000-000000000000"
with tempfile.TemporaryDirectory() as tmp:
proj = self.project(tmp, now=now)
self.teammate(
proj, sid=child_sid, name="ensign-review", stamp=stale_stamp, age=300, now=now
)
lenses = proj / child_sid / "subagents"
lenses.mkdir(parents=True)
lens_fp = lenses / "agent-lens.jsonl"
lens_fp.write_text(
json.dumps({"type": "user", "timestamp": stale_stamp, "message": {}}) + "\n"
)
(lenses / "agent-lens.meta.json").write_text(json.dumps({"name": "lens-worker"}))
os.utime(lens_fp, (now - 300, now - 300))
session = self.collect_one(tmp, None, now)

published = {a["name"]: a for a in session["subagents"]}
self.assertEqual({"ensign-review", "lens-worker"}, set(published))
self.assertIs(False, published["ensign-review"]["active"])
self.assertIs(False, published["lens-worker"]["active"])
self.assertEqual("idle", session["state"])
self.assertEqual("awaiting your message", session["state_detail"])