Skip to content

fix(persistence): preserve subagent cleanup and retention #6713

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
bolichen97 merged 1 commit into from
Sep 4, 2026
Merged

fix(persistence): preserve subagent cleanup and retention #6713

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
127 changes: 125 additions & 2 deletions docs/system-specs/modules/subagent.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -546,10 +546,133 @@ On startup, `SubagentManager` scans `~/.kiro/crew/subagents/` and reconciles:
- Created on: process death without result, delivery failure, timeout (`cause` =
`error` / `timeout` / `cancelled` / `reaped` / `gateway_restart`), **and on
successful delivery** (`cause="delivered"`, via `mark_delivered`) so `result.txt`
is retained for the grace window instead of deleted immediately.
is retained for the grace window instead of deleted immediately. The generic
writer snapshots any non-empty session ID, provider, and CWD from readable
state; live abnormal-exit values captured immediately after session acquisition
(before resume validation and context construction) override that fallback.
Cancel recovery can acquire multiple sessions under one run ID, so persistence
atomically records complete per-session cleanup generations in an owner-only
record below the file-gated `trust/` root, outside the agent-writable run folder.
Every read and write first applies the repository's fail-loud owner-only directory
restriction, including inheritable Windows DACLs, and re-locks an existing record
because tightening its parent does not retrofit an older file ACL. Only a missing
record reads as empty; I/O, parse, or schema failures propagate, so an append can
never rewrite unreadable history as a fresh one-generation record. Prune catches
those failures per tombstone and continues later entries without altering the
corrupt record; shared-session setup logs them as best-effort and keeps the live
handle instead of falling into dedicated fallback.
Each generation also records the run's retention intent and continuation owner
key before the later best-effort combined state update. Protected generation
ownership is authoritative even when readable agent-writable state supplies an
empty or conflicting key; only the referenced owner's current readable state
decides retention. The generation `keep` value is a fallback when local state
lacks that field. Session acquisition publishes the new generation
synchronously in memory before submitting durable work, so executor
saturation or cancellation cannot prevent a terminal tombstone from seeing it.
The in-memory fallback is append-only; the worker deduplicates only while
serializing the protected record off-loop and never replaces the live list, so
an older writer cannot discard a concurrent recovery SID. On the dedicated arm,
durable generation persistence follows the cancellation-drained provenance
write; both dedicated and shared identity workers are shielded and fully drained
before cancellation is re-raised, so restart cannot precede protected authority.
Cancellation cannot skip required model fields, and the already-published
memory record still feeds the terminal tombstone. Slow storage therefore cannot
stall chat/heartbeat or hide a just-acquired session from a contending tombstone,
and a transient SID1 generation-write failure cannot be lost when SID2 later
succeeds. Shared-handle ownership and provider references are attached
immediately after handle creation, before any cancellable persistence
await, so force-reap always destroys the shared handle rather than resetting a
nonexistent dedicated session. Identity persistence errors are logged without
triggering dedicated fallback or abandoning the live handle. Event-loop tombstone snapshots are memory-only: they acquire the identity
lock non-blocking and use already-published in-memory generations, never reading
the protected durable record. Executor-owned prune independently merges that
record after restart before cleanup. The protected record and in-memory fallback
are evicted when prune or explicit folder deletion succeeds. Tombstones expose
the latest identity in compatibility fields and snapshot the full list for
diagnostics/restart hints, but those agent-folder fields cannot authorize
provider deletion. Prune's deletion set comes only from the protected record
and synchronous live gateway publication, and reclaims every trusted generation.
Retention fallback selects a protected generation matching the current readable-state
SID, or the latest protected generation when the SID is absent or mismatched;
agent-writable state/tombstone SID and owner fields never select a victim identity
or suppress trusted owner/`keep` metadata.
- For readable state, only a literal current `keep is True` is retained; strings
such as `"false"` are non-retention rather than truthy policy. `true` preserves
the identity folder for restart registry rebuild; release writes `false`,
allowing prune to retry provider cleanup.
Every completed plain run records `false`; a readable legacy/failed-write
record with no key is treated as non-retained and prunes at the normal cutoff.
Readable `true` always defers disk prune; release or the conversation TTL writes
`false` and owns deletion. This arbitration is part of the cleanup fix rather
than a separate retention feature: once durable identity makes provider files
reachable, a stale `keep=False` prune racing a continuation promotion could
destroy the newly reachable resume material. Within the single gateway process,
promotion and prune arbitrate under per-agent short-held state transactions.
When a continuation tombstone points at an original owner, prune pre-resolves
that owner and acquires both per-agent locks in stable order, then re-reads under
lock; unrelated agents never contend. Promotion writes `true` before that locked
read, or prune keeps arbitration through provider cleanup and folder removal so a
later promotion returns retryable instead of racing deletion. On the event loop,
promotion probes arbitration and the per-agent off-loop-writer lock non-blocking.
Contention returns retryable `conversation_busy`, so a later retry writes
`keep=True` only after every older writer completes.
Off-loop promotion lets `update_state` acquire that non-reentrant writer lock
normally, avoiding self-deadlock while preserving serialization.
Transient persistence errors likewise return retryable without dispatch. Retry
restores the exact pre-attempt SessionManager and TTL-registry ownership; an
already-retained conversation is never unmarked. The facade returns the result
directly, so concurrent callers carry independent outcomes without a hidden
clear/call/read side channel. A process crash leaves no half-committed claim
format: the next prune re-reads the current owner state under arbitration.
If state and a rewritten tombstone both lack a top-level SID, prune derives
retention and owner from the latest valid durable cleanup generation instead of
treating the record as non-retained. Provider cleanup runs before lock release;
folder and protected-record removal follow only when every trusted generation
reports success. Unsupported providers and transient deletion failures preserve
both retry surfaces for later sweeps, capped at 90 days so a permanently missing
cleanup route cannot accumulate private run folders forever. A legacy SID present
only in agent-folder state/tombstone likewise preserves the folder inside that
window: it cannot authorize deletion, but the lookup gives a later trusted
migration time to reclaim the transcript. At the hard ceiling, only the run
folder and protected metadata are reaped; untrusted identity is never used for
provider-file deletion. Restart registry rebuild likewise accepts only literal
`keep is True`, requires the state SID and conversation owner to match a
protected/live generation, and sources provider/CWD from that trusted record;
agent-folder state cannot seed a victim SID into the later TTL release path. An
explicitly injected noncanonical state reader is an application-owned trusted
seam; the canonical disk reader never takes that compatibility fallback.
A continuation follows its original conversation's
readable `keep` value directly: `false` or a missing key is non-retention, while
unreadable owner state receives the bounded grace below instead of inheriting
the continuation's stale local `true`. Registry rebuild, prune, and TTL sweep
share `subagent_id_from_conversation_key`; malformed keys are dropped per entry
so one corrupt record cannot abort later cleanup.
- Pruned by reaper: `delivered` tombstones after `agent.subagent_result_ttl_secs`
(default 1h); all other tombstones after 7 days. `prune_stale_tombstones` takes
a per-cause cutoff for this.
a per-cause cutoff for this and treats timestamps exactly at the cutoff as
eligible, avoiding platform clock-resolution gaps. Tombstone `died` must be numeric, positive, and
non-future; string, NaN, infinity, future, and oversized values fall back to the
validated tombstone-file mtime, or the current sweep time when no valid bounded
time exists. That final fallback preserves unknown-retention grace across wall-clock
rollback instead of making the record immediately eligible.
Missing, malformed, deeply nested, or non-object tombstones are skipped for that
entry without aborting later entries in the sweep.
Missing, malformed, deeply nested, Unicode-invalid, or non-object `state.json`
is unreadable. An acquisition-time `keep=false` generation remains unknown in
this branch because a later promotion may have landed only in the now-unreadable
state; only `keep=true` may collapse uncertainty, since it can only preserve data.
A tombstone with a SID publishes only a SID-less in-memory retention hint, so the
SessionManager file-deletion exemption performs no tombstone read on the gateway
event loop without laundering agent-folder identity into provider-deletion
authority; the executor-owned restart scan
rehydrates that hint from durable tombstones before registry rebuild completes.
A readable legacy continuation whose owner
state is unreadable uses its resolved local-state SID for the same bounded
protection, even when its pre-upgrade tombstone has no SID. A malformed
continuation owner ID is likewise bounded as unknown retention for that entry;
it cannot abort processing of later tombstones. At the cutoff, unknown intent receives one extra 24-hour grace anchored on tombstone death time;
after that bounded window, trusted cleanup-generation metadata drives best-effort
provider cleanup while tombstone metadata drives folder removal eligibility.
- `spawn_status` falls back to persistence layer for completed/tombstoned agents,
reading the retained `result.txt` (and honoring offset/limit/grep).

Expand Down
125 changes: 97 additions & 28 deletions src/kiro_crew/subagent_manager/continuation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

from typing import TYPE_CHECKING

from .. import subagent_persistence as persistence
from ._component import ManagerComponent

if TYPE_CHECKING:
Expand Down Expand Up @@ -34,6 +35,7 @@
class ContinuationCoordinator(ManagerComponent):
"""Own continuation transitions while state remains facade-owned."""

_persistence = persistence
__slots__ = ()

def _conversation_busy_impl(self, conv_key: str) -> SubagentInfo | None:
Expand Down Expand Up @@ -79,37 +81,47 @@ def _conversation_busy_impl(self, conv_key: str) -> SubagentInfo | None:
return None

def _keep_recorded_on_disk_impl(self, key: str) -> bool:
"""Disk-truth continuable check for SessionManager's cache (#1115).
"""Retention guard for subagent conversations without loop-side disk probes.

True iff *key* is a subagent conversation whose run's ``state.json``
records ``keep`` — the single persisted source of retention intent.
Non-subagent keys short-circuit without touching disk.
Readable ``state.json`` remains the persisted retention authority. If
state is unreadable, the session-acquisition/tombstone path's synchronous
in-memory identity publication fails safe without reading ``tombstone.json``
on the gateway event loop.
"""
if not key.startswith("subagent:"):
conv_id = self._persistence.subagent_id_from_conversation_key(key)
if conv_id is None:
return False
conv_id = key[len("subagent:") :]
try:
state = read_state(conv_id) or {}
except Exception:
return False
return bool(state.get("keep"))
state = read_state(conv_id)
except OSError:
state = None
if isinstance(state, dict):
return state.get("keep") is True
return self._persistence.has_live_cleanup_identity(conv_id)

def _promote_conversation_impl(
self, conv_id: str, conv_key: str, last_used: float | None = None
) -> None:
"""Single choke point for promoting a conversation's retention (#1115).

Writes all three retention surfaces together so they cannot drift:
``keep=True`` in state.json (the persisted source of truth), the
SessionManager continuable cache (file-deletion exemption), and the
TTL registry entry (sweep ownership).
"""
self,
conv_id: str,
conv_key: str,
last_used: float | None = None,
) -> Any:
"""Atomically promote all retention surfaces for a conversation."""
try:
update_state(conv_id, keep=True)
except Exception:
result = self._persistence.promote_retention(conv_id, state_writer=update_state)
except OSError:
logger.debug("promote: failed to persist keep for %s", conv_id, exc_info=True)
# Preserve the established fail-safe marker for direct callers, but
# report retryable failure so continue_conversation rolls it back.
self._manager._sessions.mark_continuable(conv_key)
self._manager._conversations[conv_key] = (
last_used if last_used is not None else time.time()
)
return self._persistence.RetentionPromotionResult.RETRYABLE
if result is not self._persistence.RetentionPromotionResult.PROMOTED:
return result
self._manager._sessions.mark_continuable(conv_key)
self._manager._conversations[conv_key] = last_used if last_used is not None else time.time()
return result

def _scan_keep_states_impl(self) -> list[tuple[str, str, str, str, str, float]]:
"""Blocking scan for keep runs (#1114): read every ``state.json``
Expand All @@ -128,20 +140,46 @@ def _scan_keep_states_impl(self) -> list[tuple[str, str, str, str, str, float]]:
try:
if not d.is_dir():
continue
state = read_state(d.name) or {}
if not state.get("keep"):
state = read_state(d.name)
if not isinstance(state, dict):
tombstone = self._persistence.read_tombstone(d.name) or {}
sid = str(tombstone.get("session_id") or "")
if sid:
# Agent-folder tombstones can preserve retention/exemption
# hints, never provider-deletion authority.
self._persistence.publish_live_cleanup_hint(d.name)
continue
if state.get("keep") is not True:
continue
conv_key = str(state.get("conversation_key") or "") or f"subagent:{d.name}"
conv_id = conv_key[len("subagent:") :]
conv_id = self._persistence.subagent_id_from_conversation_key(conv_key)
if conv_id is None:
continue
sid = str(state.get("session_id") or "")
trusted_identity = self._persistence.trusted_cleanup_identity_record(
d.name,
sid,
conv_key,
)
if trusted_identity is None:
# The canonical disk reader consumes agent-writable state and
# must match protected authority. Tests/embedders may inject a
# different reader as their own trusted authority; retaining
# that established seam does not make on-disk state trusted.
if read_state is self._persistence.read_state:
continue
trusted_identity = {
"provider": state.get("provider"),
"cwd": state.get("cwd"),
}
last_used = float(state.get("updated_at") or state.get("started") or 0.0)
out.append(
(
conv_id,
conv_key,
sid,
str(state.get("provider") or PROVIDER_LABEL_DEFAULT),
str(state.get("cwd") or ""),
str(trusted_identity.get("provider") or PROVIDER_LABEL_DEFAULT),
str(trusted_identity.get("cwd") or ""),
last_used,
)
)
Expand Down Expand Up @@ -301,7 +339,34 @@ def continue_conversation_impl(
# (#1115): state.json keep=True (tombstone pruner skips deletion),
# the SessionManager continuable cache, and the TTL registry entry.
# The conversation TTL sweep / spawn_release owns deletion from here.
self._manager._promote_conversation(conv_id, conv_key)
# Snapshot existing ownership: a retryable promotion attempt must undo
# only state it introduced, never erase an earlier keep/continuation.
was_continuable = self._manager._sessions.is_continuable(conv_key)
had_previous_last_used = conv_key in self._manager._conversations
previous_last_used = self._manager._conversations.get(conv_key, 0.0)
# The facade forwards the enum result directly. Legacy tests and external
# monkeypatches that return None/non-enum retain the historical promoted
# behavior; only an explicit RETRYABLE outcome alters dispatch.
promotion = self._manager._promote_conversation( # type: ignore[func-returns-value]
conv_id, conv_key
)
if promotion is self._persistence.RetentionPromotionResult.RETRYABLE:
if not was_continuable:
self._manager._sessions.unmark_continuable(conv_key)
if not had_previous_last_used:
self._manager._conversations.pop(conv_key, None)
else:
self._manager._conversations[conv_key] = previous_last_used
return SubagentInfo(
id=uuid.uuid4().hex[:8],
task=_redact(task),
done=True,
parent_session_key=parent_session_key,
error=(
"conversation_busy: retention promotion is temporarily "
f"unavailable for {conv_id}; retry the continuation"
),
)
inc_memory, inc_lessons, inc_project = self._manager._inherited_context_groups(conv_id)
# A continuation has to run WHERE THE RUN RAN. `spawn` resolves an empty
# cwd to the pool project before it validates the agent name, so a run
Expand Down Expand Up @@ -734,7 +799,11 @@ def _sweep_conversations_impl(self, now: float) -> None:
if self._manager._conversation_busy(conv_key) is not None:
self._manager._conversations[conv_key] = now # active — refresh
continue
conv_id = conv_key[len("subagent:") :]
conv_id = self._persistence.subagent_id_from_conversation_key(conv_key)
if conv_id is None:
logger.warning("Dropping malformed conversation registry key %r", conv_key)
self._manager._conversations.pop(conv_key, None)
continue
ok, detail = self._manager.release_conversation(conv_id)
logger.info(
"Conversation %s expired after %ds idle: %s",
Expand Down
Loading
Loading