SWF v41 baseline: episodes engine - #45
Merged
Merged
Conversation
EpisodeDefinition (the per-workflow contract: message-to-event mapping, participant recognition, end signal, completion pass), EpisodeBuilder (routes bus traffic to armed definitions and drives open, append, completion, close per execution; never raises into the listening agent), and MonitorEpisodeIngest (REST client for the monitor episode ingest endpoints). A workflow gains an episode record by implementing one definition; per snapper-ai docs/EPISODES.md and swf-testbed docs/agentic-workflow-view.md.
A per-episode seen set keeps steady message traffic from re-upserting its sender on every message; death reports pass through.
started_at and ended_at hooks let a definition supply normalized message times in place of arrival times — the close carries the recorded end, which backfilled episodes require.
adopt_open_episodes resumes the builder identity's open episodes at startup — an episode whose end signal already passed is driven to completion and close, one still mid-flight keeps appending — so a builder restart never orphans a live record. The ingest client gains the episode list and detail reads this needs, and every handled message carries an arrival stamp as the fallback event time for writers that stamp nothing.
A send that fails now marks the connection down, attempts one reconnect and resend regardless of error type (the old error-string filter missed NotConnectedException), and raises if the message is still unsent. The silent swallow let a dying agent's workflow run to a false 'completed' while its end_run evaporated, abandoning RunState 102780 (2026-07-30) as non-terminal — the System page stale-state warning. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Introduces the v41 “episodes” baseline by adding a workflow-agnostic episode-building engine (definitions, builder, and monitor ingest client), and updates agent/background-execution documentation plus MQ send semantics.
Changes:
- Added
episodes.pyimplementingEpisodeDefinition,EpisodeBuilder,EpisodeContext, andMonitorEpisodeIngestfor swf-monitor episode ingest. - Updated
BaseAgent.send_messageto retry once on any send failure and raise on unrecoverable failure. - Expanded concurrency migration guidance for
run_in_backgroundin both code docs and README.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 4 comments.
| File | Description |
|---|---|
| src/swf_common_lib/episodes.py | Adds the core episodes engine and monitor ingest client used by workflow-specific episode definitions/builders. |
| src/swf_common_lib/base_agent.py | Documents background worker concurrency semantics and changes send_message to raise after failed reconnect+retry. |
| README.md | Adds detailed “safe migration” guidance for background-worker concurrency and locking patterns. |
Suppressed comments (2)
src/swf_common_lib/episodes.py:105
_get()returnsresponse.json()directly; a non-JSON response will raiseValueErrorand bypass theEpisodeIngestErrorhandling, potentially escaping out ofEpisodeBuildermethods. Catch JSON decode errors and rethrow asEpisodeIngestError.
return response.json()
src/swf_common_lib/episodes.py:298
participants_from_message()is allowed to return entries without anid, buthandle_message()will addNonetoseen_participantsand may incorrectly suppress future upserts. Filter out entries missingidand only add non-empty ids toseen_participants.
participants = [
entry for entry in definition.participants_from_message(message)
if not (entry.get("id") in context.seen_participants
and "died_at" not in entry)
]
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+64
to
+69
| if response.status_code >= 400: | ||
| raise EpisodeIngestError( | ||
| f"POST {url} returned {response.status_code}: " | ||
| f"{response.text[:500]}" | ||
| ) | ||
| return response.json() |
Comment on lines
+312
to
+315
| except EpisodeIngestError as exc: | ||
| logger.error("episode ingest failed for %s: %s", | ||
| execution_id, exc) | ||
| return False |
Comment on lines
+337
to
+347
| try: | ||
| self.ingest.close( | ||
| scope=definition.scope, | ||
| episode_id=execution_id, | ||
| ended_at=context.ended_at or context.end_seen_at, | ||
| summary=definition.summary(context), | ||
| ) | ||
| except EpisodeIngestError as exc: | ||
| logger.error("episode close failed for %s: %s", | ||
| execution_id, exc) | ||
| del self.active[execution_id] |
Comment on lines
+254
to
+258
| for event in record.get("events", []): | ||
| context.seen_participants.add(event.get("participant")) | ||
| if definition.is_end({"msg_type": event.get("kind")}): | ||
| context.end_seen_at = utc_now_iso() | ||
| context.ended_at = event.get("time") |
An episode with no end seen has no deadline to pass; the caller's guard made this unreachable, and the narrowing states it where mypy checks. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Test Coverage Summary |
wenaus
added a commit
that referenced
this pull request
Aug 8, 2026
Copilot review findings on PR #45, verified: ingest responses that decode as non-JSON now raise EpisodeIngestError instead of escaping the callers' catch; handle_message() traps definition-hook failures so one malformed bus message cannot take down the listening agent; tick() retains an episode for retry when its close fails and falls back to an empty summary when the summary hook fails. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Baseline PR for the v41 cycle. The episodes module: a generic episode-building engine with definition hooks for start and end times, builder adoption, ingest reads, arrival stamps, and single reporting per participant. send_message raises on unrecoverable failure instead of swallowing it.
Canonical release record: RELEASE_NOTES.md v41 entry in swf-testbed PR BNLNPPS/swf-testbed#69
Companion: swf-monitor PR BNLNPPS/swf-monitor#46
🤖 Generated with Claude Code