SWF v41 baseline: workflow episodes, fast processing v11, release notes - #69
Merged
Conversation
The Fast Processing Agent consumes stf_ready and makes TF slices directly from arriving STFs; the FastMon Agent and its STF sample stage are no longer part of the fast processing chain. The workflow doc now embeds the current diagram (it embedded v5). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The Fast Processing Agent consumes stf_ready and makes TF slices directly from arriving STFs; the FastMon Agent and STF sample stage are out of the fast processing chain. Message tables, sequence diagram, data products, config (slices_per_stf), and the runbook examples state the simplified flow. The running agent still implements the prior sampled chain; code follows the doc. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
pg_isready is not on PATH here and unprivileged lsof cannot see other users' listening sockets, so both checks failed on healthy services. A socket connect needs neither and tests what matters.
Enable the data agent that registers runs and STF files from epictopic and feeds the processing queue.
Point the default fast-processing configuration at torre1 so stale environment overrides cannot recreate or mislabel agents as torre2.
Temporal view of one workflow execution: dynamic lanes for agents and PanDA-job workers, message marks, live and replay modes. Records the 2026-08-02 evidence audit (execution reconstructable today from existing records), the data contract, the writer gaps to fill (consumption records, processing-agent announcements, exit stamps, durable episode capture), and the delivery phases through production scale.
The episodes package: the shared testbed message-to-event mapping, the prompt_processing definition with its PanDA task and job completion join, and the stf_datataking definition. The episode builder agent listens on the epictopic without namespace filtering (episodes are scope-wide), drives the definitions through the swf_common_lib.episodes engine, and registers as a supervisord program (autostart off). Per docs/agentic-workflow-view.md.
…n; backfill The episode builder agent is exempt from the workflow agent lifecycle: STANDING_PROGRAMS excludes it from the running-agents check, stop-agents leaves it up, autostart revives it through the run flow's supervisord restart, and that restart now waits for the previous instance to be provably gone instead of assuming a one second shutdown. Testbed bus timestamps are naive local stamps; the shared mapping attaches the agents' zone and supplies the episode start and end hooks. episodes/backfill.py replays a past execution's recorded messages through the same definitions and engine, joining PanDA to closure with recorded times — first exercised on prompt_processing-zyang2-0845 (run 102827: 114 events, 34 participants including 30 PanDA-job worker lanes).
The builder identity is the stable 'episode-builder' so restarts resume open episodes; the supervisord program is retired in favor of the hand-installed swf-episode-builder systemd unit (runs as the operating account, autostarts at boot, Restart=always; the unit carries the CA bundle paths because the ~/.env loader skips shell-computed lines). Message times fall back through recorded sent time and arrival stamp for writers that stamp nothing — the backfill injects the recorded sent time. STANDING_PROGRAMS empties but the mechanism stays.
A stop request during a workflow returned through the normal completion path: the execution read 'completed' though end_run was never sent, leaving its announced run states claiming activity. The stop exit now marks the execution 'terminated' and abandons its announced runs; the completed-process check moves ahead of the stop check so a finished workflow always reads 'completed' when a stop races its last event. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
After swf-monitor PR #45 the slice record references its registered TF sample (FastMonFile) instead of carrying flat tf_filename/stf_filename strings. The REST create payload changes accordingly; the worker queue message is unchanged. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This was referenced Aug 7, 2026
Member
Author
|
Companion PRs: swf-monitor BNLNPPS/swf-monitor#46 · swf-common-lib BNLNPPS/swf-common-lib#45 |
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Baseline v41 follow-up for swf-testbed, adding workflow-episode capture plumbing (definitions + builder/backfill tooling), aligning fast-processing documentation/agent behavior with the v11 pipeline contract, and tightening run/execution lifecycle handling on mid-run stops.
Changes:
- Terminalizes executions (and abandons announced runs) when a workflow is stopped mid-run, while preserving “completed” on true completion.
- Adds episode definitions + common message mapping, plus an episode builder agent and a backfill utility for recorded message replays.
- Updates CLI service probing to use direct TCP connects (instead of shell tooling) and adjusts agent stop behavior to avoid stopping standing programs.
Reviewed changes
Copilot reviewed 18 out of 19 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| workflows/workflow_runner.py | Returns explicit workflow outcomes and marks executions terminated on mid-run stop, abandoning announced runs. |
| workflows/testbed.toml | Enables the data agent by default with explanatory comments. |
| workflows/orchestrator.py | Introduces STANDING_PROGRAMS and filters standing programs out of “running agents” detection; improves supervisord restart shutdown wait. |
| workflows/fast_processing_default.toml | Aligns namespace default with the main testbed config. |
| src/swf_testbed_cli/main.py | Reworks Postgres/ActiveMQ checks to TCP probes; updates stop-agents to stop only workflow agents. |
| RELEASE_NOTES.md | Adds v41 release notes entry. |
| example_agents/fast_processing_agent.py | Updates TF-slice POST payload to reference the registered TF sample (fastmon_file) and omit filename fields. |
| episodes/common.py | Adds shared message→event/participant mapping for episode capture. |
| episodes/prompt_processing.py | Adds prompt-processing episode definition with PanDA join in completion pass. |
| episodes/stf_datataking.py | Adds STF datataking episode definition. |
| episodes/backfill.py | Adds script to backfill episodes from recorded workflow messages. |
| episodes/init.py | Registers available episode definitions. |
| docs/prompt-processing-workflow.md | Documents background execution guidance and safe worker-count settings. |
| docs/images/fast-processing-pipeline-v11.svg | Adds updated v11 pipeline diagram asset. |
| docs/fast-processing-workflow.md | Updates workflow docs to match v11 (remove FastMon stage, rename messages/products). |
| docs/architecture_and_design_choices.md | Adds explicit guidance on responsiveness vs concurrency in background execution. |
| docs/agentic-workflow-view.md | Adds design doc for an agentic execution replay/live view backed by episode capture. |
| agents/episode_builder_agent.py | Adds an episode builder agent that listens on /topic/epictopic and ingests episodes into the monitor. |
| agents.supervisord.conf | Trailing whitespace/newline change only. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+38
to
+45
| for field in ('timestamp', 'sent_at', '_received_at'): | ||
| raw = message.get(field) | ||
| if not raw: | ||
| continue | ||
| try: | ||
| parsed = datetime.fromisoformat(str(raw)) | ||
| except ValueError: | ||
| continue |
Comment on lines
+109
to
122
| def _check_tcp_service(name, host, port): | ||
| """Probe a TCP service by connecting to it. External tools (pg_isready, | ||
| lsof) are unreliable here: not on PATH, or blind to other users' | ||
| sockets without privileges. A connect() needs neither.""" | ||
| print(f"--- Checking {name} connection at {host}:{port} ---") | ||
| try: | ||
| result = subprocess.run( | ||
| ["pg_isready", "-h", db_host, "-p", db_port, "-U", db_user, "-d", db_name], | ||
| capture_output=True, | ||
| text=True, | ||
| check=True, | ||
| ) | ||
| print(result.stdout.strip()) | ||
| if "accepting connections" not in result.stdout: | ||
| print("Warning: PostgreSQL is not ready.") | ||
| return False | ||
| with socket.create_connection((host, int(port)), timeout=3): | ||
| pass | ||
| print(f"{name} is accepting connections at {host}:{port}.") | ||
| return True | ||
| except (subprocess.CalledProcessError, FileNotFoundError) as e: | ||
| print(f"Error checking PostgreSQL status: {e}") | ||
| print("Please ensure PostgreSQL is running and `pg_isready` is in your PATH.") | ||
| except OSError as e: | ||
| print(f"Warning: could not connect to {name} at {host}:{port}: {e}") | ||
| print(f"Please ensure {name} is running.") | ||
| return False |
Comment on lines
+328
to
+333
| # Stop workflow agents; standing infrastructure programs (the | ||
| # episode builder) are not part of any workflow's lifecycle and | ||
| # stay up. | ||
| sys.path.insert(0, str(testbed_root)) | ||
| from workflows.orchestrator import STANDING_PROGRAMS, get_running_agents | ||
|
|
wenaus
added a commit
that referenced
this pull request
Aug 8, 2026
_check_tcp_service() warns instead of aborting when a port env var is not an integer; drop an unused import. 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.
Follow-up baseline PR for the v41 cycle (PR #65 merged the early portion). This release introduces workflow episodes as a recorded unit (episode definitions and a builder agent running as a standing system service), adopts the fast processing pipeline v11 design of record with the example agent updated to the restructured TF slice REST contract, terminalizes executions and announced runs on mid-run stop, and folds in contributed data_agent, prompt_processing, and CI work.
Canonical release record: RELEASE_NOTES.md v41 entry in this PR.
Companion PRs: swf-monitor and swf-common-lib baseline PRs (linked in comments).
🤖 Generated with Claude Code