diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index c03df3e..18215b1 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -1,5 +1,119 @@ # Release Notes +## v41 (2026-08-07) + +This release extends Snapper with site-level job histories and a campaign delivery view, introduces workflow +episodes as a recorded unit across the stack, adds verified PanDA task pause and resume with paced bulk +controls, restructures the TF slice record around its registered TF sample, and moves the heavy monitor pages +to cached serving. The epicprod domain gains the delivered-data daily record and the campaign plan curation +surface. These notes cover the coordinated baseline repositories and the relevant `main` work in their peer +repositories. + +### Snapper: site job histories and campaign delivery (snapper-ai, swf-monitor) + +- Site-level job lifecycle histories: the PanDA component publishes cumulative terminal-outcome counters with + retrofill, and a Site focus renders per-site job curves in lifecycle order with application-standard state + colors. The site slice is a table with one row per curve; site facts drill down to the jobs pages, and + drilldowns are limited to persistent outcomes. The jobs-page graphics align with the selected Snapper + interval and use consistent semantic colors for failure classes. +- Campaign delivery view: per-campaign tabs with delivery lens curves, a per-PC daily arrivals quilt, a + day-breakdown cut card covering the selected campaigns, and events and files as toggled quantities, served + from cached series. Delivery leaves account by progress rule, one leaf per configuration. +- URL state: curve tick selection, provider-owned default-off curves, live-edge (`cut=now`) tracking URLs, and + sticky chosen cuts; display defaults never stamp the URL. Focus-sized series caching follows refreshes + promptly. +- The numbered activity bars open run stories with machine-readable failed-file counts and application-standard + state fills; low-information curves are removed from the job and task plots. + +### Workflow episodes (swf-common-lib, snapper-ai, swf-monitor, swf-testbed) + +- The swf-common-lib episodes module is a generic episode-building engine: definitions specify event-fed + bounded records with start and end hooks; the builder adopts events, stamps arrivals, and reports each + participant once. +- The testbed episode builder runs as a standing system service, capturing workflow episodes off the message + bus for all namespaces. +- swf-monitor provides episode REST ingest and read endpoints backed by the snapper-ai episode store, and the + Snapper Time history adds an Episode view. An agentic workflow view over these records is planned, with plan + documents in swf-testbed and snapper-ai. + +### PanDA task operations and Capcom (swf-monitor, swf-epicprod) + +- Verified pause and resume for PanDA tasks: operator controls restricted to authorized users, executed and + verified against PanDA, with paced bulk operations rendered above the dynamic task table. An alarm-driven + automatic pause on failure-rate spikes is planned (swf-epicprod). +- Capcom is an operations communication console: a page of state tiles showing current SWF operational state, + alongside a feed of operational notices. The console itself is currently external to SWF, implemented in a + separate operations system, and will probably be added to SWF. swf-monitor serves it: a state endpoint + provides tile-exact SWF state; alarm presence and the swf-bot dispatcher are state tiles; a nightly + campaign-delivery notice reports the day's arrivals; notices are buffered for consumer polling instead of + external push; the PanDA state includes the running task count and integer success percent. + +### Fast processing chain and TF slice restructure (swf-monitor, swf-testbed) + +- The fast processing pipeline v11 is the design of record: STF files slice directly and the separate FastMon + processing stage is retired; the TF sample persists as a registration record (FastMonFile) that parents the + slices. +- The TFSlice record is keyed to its registered TF sample by foreign key and the flat tf_filename/stf_filename + columns are removed (monitor_app migrations 0005 and 0006). REST and MCP outputs are unchanged. The REST + create payload replaces the two filename fields with `fastmon_file` naming the registered sample; clients + that create slices must be updated accordingly. The example fast processing agent posts the new payload; its + worker queue messages are unchanged. +- The fastmon-files list page 500 (unbound short_filename) is fixed. + +### PCS: delivered-data record, campaign plan, physics configuration (swf-epicprod, swf-monitor) + +- The campaign delivery daily record measures per-file events and accumulates per-PC arrivals over complete + days, backfilled from Rucio arrival history and produced nightly in the catalog_sync chain, bounded to + active and recorded campaigns. CAMPAIGN_DELIVERY.md documents the record and its views. +- Campaign plan: a target-events curation surface (/pcs/plan/) with expected-events fields, a setter service, + and a REST endpoint; request-material target derivation; campaign pages link the campaign's delivery time + history. +- The physics configuration is an entity (PhysicsConfig) and is documented as such. The physics page and + request composer serve cached products; an unknown tag type is a 404. +- Catalog: forward-only rotation with server-side guards; the latest daily AI assessment appears inline on + current and producing campaign pages; assessment strips name their campaign and use human-addressed URLs. +- EPICPROD_VALIDATION records the agreed production–validation loop. Assessment causal claims require evidence + and structured attribution. + +### Serving performance and deployment (swf-monitor) + +- The PanDA activity page, tasks list, compute-usage rollup, and log summary serve from cached products; the + tasks window cache renders cells at serve time. Typed REST filters reject malformed values with 400. + django-filter is enabled so filterset_fields declarations filter. +- Deploys stage a unique release and never rebuild the current tree in place; the Apache maximum-requests + daemon recycling is removed. +- The Mattermost bots keep a stable prompt-cache prefix and log usage per round. + +### Interface work (swf-monitor) + +- List pages: enumerated value walls are removed as filters; long names render middle-elided with the full + name on hover; the TF slices page is reorganized; latching table headers on all list pages with adaptive + two-axis scrolling for wide tables; visible checkbox borders application-wide. +- The account page shows testbed activity lanes on the My Workflows tab; the testbed home places Snapper + namespace lanes above Workflows; hub routes are authoritative for navigation mode; robots.txt is served + outside the OIDC gate and disallows all crawling. + +### External access (swf-remote, site-canary) + +- swf-remote: crawlers are denied the whole proxied surface at the proxy choke point; a live-data access + contract with proxy identities; the aggregate System page is proxied; GitHub sign-in; a rolling session + window. +- site-canary: queue assessments state their evaluation window, keep status evidence-current, and link to + Snapper and source documentation; Canary queues publish into Snapper site state with updated state colors. + +### Agents and CI (swf-testbed) + +- data_agent supports uploading to deterministic RSEs; prompt_processing adds a configurable mode for real + file processing; the integration test runs several workflows. +- Run lifecycle: executions and announced runs terminalize on mid-run stop; status checks probe TCP directly + instead of shelling out; the testbed image installs site-canary before the meta-package. + +### Acknowledgments + +- Wen Guan: the TFSlice restructure (swf-monitor PR #45) and the fast processing workflow it serves. +- Dmitry Kalinkin: data agent deterministic-RSE uploads, prompt processing real-file mode, and the + multi-workflow integration test (swf-testbed PRs #68, #66, #63). + ## v40 (2026-07-25) This release adds Snapper, which records subsystem-published operational state as a queryable history, with a diff --git a/agents.supervisord.conf b/agents.supervisord.conf index 615434c..ba43769 100644 --- a/agents.supervisord.conf +++ b/agents.supervisord.conf @@ -103,3 +103,4 @@ stopwaitsecs=10 stopsignal=QUIT stdout_logfile=%(here)s/logs/%(program_name)s.log stderr_logfile=%(here)s/logs/%(program_name)s.log + diff --git a/agents/episode_builder_agent.py b/agents/episode_builder_agent.py new file mode 100644 index 0000000..c719ada --- /dev/null +++ b/agents/episode_builder_agent.py @@ -0,0 +1,64 @@ +#!/usr/bin/env python3 +"""Episode builder agent: records workflow episodes from bus traffic. + +One more listener on the epictopic, in keeping with the open-listening +messaging philosophy: it consumes every message, routes executions to +the armed episode definitions (the episodes package), and drives each +episode through open, append, completion, and close against the +monitor's episode ingest REST (docs/agentic-workflow-view.md). + +Episodes are scope-wide, so unlike workflow agents this agent applies +no namespace filtering: every namespace's executions are recorded. +""" + +import json +import logging +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from swf_common_lib.base_agent import BaseAgent +from swf_common_lib.episodes import EpisodeBuilder, MonitorEpisodeIngest + +from episodes import ALL_DEFINITIONS + + +class EpisodeBuilderAgent(BaseAgent): + def __init__(self): + super().__init__(agent_type='EPISODE_BUILDER', + subscription_queue='/topic/epictopic') + # The builder identity is stable across restarts and instances: + # a restarted builder must resume the episodes its predecessor + # opened. The per-process agent name would strand them. + ingest = MonitorEpisodeIngest( + base_url=self.base_url, + token=self.api_token, + builder_identity='episode-builder', + ) + self.builder = EpisodeBuilder( + [definition() for definition in ALL_DEFINITIONS], ingest) + logging.info('armed definitions: %s', + [d.workflow_name for d in self.builder.definitions]) + adopted = self.builder.adopt_open_episodes() + if adopted: + logging.info('adopted %d open episode(s) from a previous ' + 'builder instance', adopted) + + def on_message(self, frame): + try: + message = json.loads(frame.body) + except (ValueError, TypeError) as exc: + logging.error('unparseable message dropped: %s', exc) + return + self.builder.handle_message(message) + + def send_heartbeat(self): + result = super().send_heartbeat() + # The heartbeat cycle is the completion-poll cadence. + self.builder.tick() + return result + + +if __name__ == '__main__': + EpisodeBuilderAgent().run() diff --git a/docs/agentic-workflow-view.md b/docs/agentic-workflow-view.md new file mode 100644 index 0000000..07c2ca7 --- /dev/null +++ b/docs/agentic-workflow-view.md @@ -0,0 +1,124 @@ +# Agentic workflow view + +A temporal view of one workflow execution: the agents that ran it, the +workers it provisioned, the messages that passed between them, and the +data that flowed through them — rendered as a live display while the +workflow runs and as a replay of any past execution. The view is a +Snapper surface; the engine mechanisms it requires are specified in +snapper-ai `docs/EPISODES.md`. + +## Display + +The plot is a stack of horizontal lanes over a time axis, in the +established Snapper Time history vocabulary (lanes, tiles, cut, +floater, step arrows). + +- One lane per participant. The workflow runner and the agents open + the stack; worker lanes appear below as workers come into + existence. A prompt-processing execution opens with a handful of + lanes and fans out to tens of worker lanes, then converges as jobs + finish and the run tears down. +- Lanes are dynamic: a lane begins when its participant registers and + ends when it exits. A finished participant's lane keeps its + vertical slot, dimmed, to the end of the episode, so the fanout and + convergence render as a stable shape and the vertical layout never + reshuffles during a replay. +- Messages render as marks at their send time on the sender's lane, + with connectors to recorded consumers where consumption records + exist (see Data contract below). +- Activity on a lane renders as tiles: an agent's processing spans, a + worker's created / started / finished phases. +- A click on any element opens its detail card: message payload, + agent record, PanDA job record, file record. +- The cut and step arrows carry their Time history meanings: a click + is a time slice, the arrows step the window. An execution-stepping + mode — arrows move between executions rather than time windows — + is a candidate addition. +- Live mode: while the execution runs, the view follows it, either + pseudo-realtime as events land or promptly after completion. Both + are feasible with current record latencies (seconds for messages + and file records, minutes for PanDA job state). + +## Evidence audit + +Audited 2026-08-02 against prompt-processing execution +`prompt_processing-zyang2-0845` (run 102827, 15 STFs, decision-box +broadcast to E1_BNL and E1_JLAB) and the stf_datataking executions of +the same day. The full event sequence is reconstructable from +existing records, all in the system database with sub-second +timestamps: + +| time (UTC) | event | record | +|---|---|---| +| 20:00:13 | execution starts | WorkflowExecution | +| 20:00:18 | run_imminent | message log | +| 20:00:26 | start_run | message log | +| 20:00:27–20:01:02 | stf_gen ×15, pause/resume around standby | message log | +| 20:00:29.6/.7 | stf_ready per site: the decision-box fanout | message log | +| 20:00:34.6 | PanDA tasks created, one per site | PanDA tasks | +| 20:05:55 | 15+15 jobs created | PanDA jobs | +| 20:11:22–43 | jobs start | PanDA jobs | +| 20:13–20:14:43 | jobs finish, tasks done | PanDA jobs, tasks | + +## Data contract + +Sources, all local database reads (no remote calls in the render +path): + +- **WorkflowExecution** — episode identity, start and end, full + parameter set including agent roster and workflow configuration. +- **WorkflowMessage** — sender agent, type, namespace, execution id, + run id, payload, sent-at to the microsecond. +- **SystemAgent** — lane birth (`created_at` at registration), lane + death (`operational_state` EXITED, stamped by `updated_at`), + heartbeats, pid, hostname. +- **STF files / TF slices** — file-level flow; TF slice records name + their `assigned_worker`. +- **PanDA tasks** — the run number is embedded in the task name + (`user..swf..processed./`), giving a direct join + from execution to tasks; creation, start, and end times per task. +- **PanDA jobs** — one record per worker with creation, start, and + end times, site, and output metadata; jobs join to tasks by + `jeditaskid`. For these workflows the worker lanes are the PanDA + jobs born within the execution's tasks. + +## Gaps to fill + +Verified in code at the executing commits, 2026-08-02: + +1. **Consumption records.** Messages record their sender only. The + messaging philosophy is open listening — any agent may subscribe — + so an addressee list recorded at send time would misstate the + model. Instead, consumers record consumption: an agent that receives a + message and acts on it records that fact with message id and + timestamp. This yields the message connectors for the view and, + independently, a workflow-integrity tool: a message no agent + consumed, or a message consumed by an unexpected agent, becomes a + detectable condition. Requires team discussion before + implementation. +2. **Processing agent announcements.** The prompt-processing agent + emits only heartbeats; its PanDA task submissions are visible only + through the task records appearing. It should announce submission + events on the bus like its peers announce theirs. +3. **Explicit agent exit stamp.** Lane death is currently inferred + from the EXITED transition's `updated_at`; an explicit exit + timestamp would remove the approximation. +4. **Durable episode capture.** Messages and file records are + operational logs with retention policies. An episode is captured + into a durable Snapper record at (or promptly after) execution + end, so replay never depends on raw log retention. + +## Delivery phases + +1. **Episode capture** — a builder that joins the sources above into + a durable episode record for each execution, live or promptly + after completion. +2. **Replay view** — the lane display for any captured episode. +3. **Live mode** — the view follows a running execution. +4. **Production scale** — the same mechanism applied to epicprod + workflows: campaign tasks fanning out across grid sites, with + PanDA jobs as worker lanes. The mechanism is generic; only the + provider data differs. + +Related documentation: `fast-processing-workflow.md`, +`e0-e1-state-machine.md`, snapper-ai `docs/EPISODES.md`. diff --git a/docs/architecture_and_design_choices.md b/docs/architecture_and_design_choices.md index 1fa9de6..97259b5 100644 --- a/docs/architecture_and_design_choices.md +++ b/docs/architecture_and_design_choices.md @@ -80,6 +80,17 @@ contributors and for future architectural reviews. `dedup_key` to avoid the duplicate-work race concurrency introduces; a send lock makes worker-thread sends safe and shutdown drains the pool. +- **Concurrency is separate from responsiveness:** the bounded pool uses + `SWF_AGENT_MAX_WORKERS` and defaults to 4, so calls with different + `dedup_key`s may execute at the same time. Deduplication is not serialization. + The safe first migration of an existing serial handler is + `SWF_AGENT_MAX_WORKERS=1`: work leaves the receiver thread, but doers still run + one at a time. Use more workers only after per-message state has moved out of + shared `self.*` fields and every shared library, client, temporary path, and + process-global operation has been audited. A non-thread-safe subsection may + instead use a dedicated lock acquired inside the background doer; never wait + for that lock in `on_message`. + - **API and consumers:** the API is documented in the `swf-common-lib` README; the first consumer is the epicprod ops agent (`swf-epicprod/docs/EPICPROD_OPS_AGENT.md`). diff --git a/docs/fast-processing-workflow.md b/docs/fast-processing-workflow.md index 036a07c..708b693 100644 --- a/docs/fast-processing-workflow.md +++ b/docs/fast-processing-workflow.md @@ -6,13 +6,12 @@ This document describes the fast processing workflow for near real-time detector The fast processing workflow enables rapid processing of detector data by: 1. Simulating DAQ data taking (STF generation) -2. Sampling Time Frames (TF) from Super Time Frames (STF) -3. Creating TF slices for parallel processing -4. Distributing slices to PanDA workers running reconstruction payloads +2. Creating TF slices directly from arriving STFs for parallel processing +3. Distributing slices to PanDA workers running reconstruction payloads ### Pipeline Overview -![Fast Processing Pipeline](images/fast-processing-pipeline-v5.svg) +![Fast Processing Pipeline](images/fast-processing-pipeline-v11.svg) **Worker Payload:** Each PanDA worker receives a TF slice as input and runs a reconstruction payload. Currently the payload is a placeholder; in production it will be EICrecon for ePIC detector reconstruction. @@ -24,13 +23,11 @@ The fast processing workflow enables rapid processing of detector data by: flowchart LR DS["DAQ Simulator"] DA["Data Agent"] - FM["FastMon Agent"] FP["Fast Processing Agent"] W["PanDA Workers
(EICrecon)"] DS -->|"stf_gen
STF"| DA - DA -->|"stf_ready"| FM - FM -->|"tf_file_registered
STF sample"| FP + DA -->|"stf_ready"| FP FP -->|"TF slices"| W ``` @@ -40,7 +37,6 @@ flowchart LR sequenceDiagram participant DS as DAQ Simulator participant DA as Data Agent - participant FM as FastMon Agent participant FP as Fast Processing Agent participant PQ as PanDA Queue participant MON as Monitor DB @@ -48,32 +44,24 @@ sequenceDiagram Note over DS,MON: === Run Initialization === DS->>DA: run_imminent (execution_id, run_id) - DS->>FM: run_imminent DS->>FP: run_imminent DA->>MON: Create dataset for run FP->>MON: Initialize RunState DS->>DA: start_run (run_id) - DS->>FM: start_run DS->>FP: start_run FP->>MON: Update RunState (phase=physics) Note over DS,MON: === STF Processing (repeat for each STF) === DS->>DA: stf_gen (filename, sequence) - DS->>FM: stf_gen DS->>FP: stf_gen DA->>MON: Register STF file DA->>DA: Process STF metadata - DA->>FM: stf_ready (filename, run_id) - DA->>FP: stf_ready + DA->>FP: stf_ready (filename, run_id) - FM->>FM: Sample TFs from STF - FM->>MON: Register TF samples - FM->>FP: tf_file_registered (tf_filename, stf_filename) - - FP->>FP: Create TF slices + FP->>FP: Create TF slices from the STF FP->>MON: Register TF slices loop For each slice @@ -85,7 +73,6 @@ sequenceDiagram Note over DS,MON: === Run Completion === DS->>DA: end_run (total_stf_files) - DS->>FM: end_run DS->>FP: end_run DA->>MON: Finalize dataset @@ -103,15 +90,10 @@ Run (run_id: 101993) │ ├── swf.101993.000002.stf │ └── swf.101993.000003.stf │ -├── TF Samples (Time Frame samples from FastMon) -│ ├── swf.101993.000001_tf_001.tf -│ ├── swf.101993.000001_tf_002.tf -│ └── ... -│ └── TF Slices (for PanDA workers) ├── swf.101993.000001_slice_000.tf ├── swf.101993.000001_slice_001.tf - └── ... (15 slices per STF sample) + └── ... (15 slices per STF) ``` ### Data Product Details @@ -119,7 +101,6 @@ Run (run_id: 101993) | Product | Created By | Stored In | Purpose | |---------|-----------|-----------|---------| | **STF File** | DAQ Simulator | STFFile table | Raw detector data unit | -| **TF Sample** | FastMon Agent | FastMonFile table | Sampled subset for fast monitoring | | **TF Slice** | Fast Processing Agent | TFSlice table | Processing unit for PanDA workers | ## Message Types @@ -139,14 +120,13 @@ Run (run_id: 101993) | Message | From | To | Payload | |---------|------|----|---------| -| `stf_ready` | Data Agent | FastMon, Fast Processing | `filename`, `checksum`, `size_bytes` | -| `tf_file_registered` | FastMon Agent | Fast Processing Agent | `tf_filename`, `stf_filename`, `tf_count` | +| `stf_ready` | Data Agent | Fast Processing Agent | `filename`, `checksum`, `size_bytes` | ### Queue Messages (Fast Processing → PanDA) | Message | Destination | Payload | |---------|-------------|---------| -| `slice` | `/queue/panda.transformer.slices` | `slice_id`, `tf_filename`, `start`, `end`, `tf_count` | +| `slice` | `/queue/panda.transformer.slices` | `slice_id`, `stf_filename`, `start`, `end`, `tf_count` | ## Configuration @@ -159,9 +139,6 @@ namespace = "torre2" [agents.data] enabled = true -[agents.fastmon] -enabled = true - [agents.fast_processing] enabled = true @@ -169,8 +146,7 @@ enabled = true stf_count = 10 # STF files to generate physics_period_count = 1 # Physics periods per run target_worker_count = 30 # Target PanDA workers -stf_sampling_rate = 1.0 # Fraction of STFs to sample (1.0 = 100%) -slices_per_sample = 15 # TF slices per STF sample +slices_per_stf = 15 # TF slices per STF slice_processing_time = 30 # Seconds per slice (for planning) ``` @@ -247,7 +223,7 @@ stateDiagram-v2 | Metric | Source | Purpose | |--------|--------|---------| | `stf_count` | WorkflowExecution | Total STFs in run | -| `tf_files_received` | Fast Processing Agent | TF samples processed | +| `stfs_received` | Fast Processing Agent | STFs processed | | `slices_created` | RunState | Total slices generated | | `slices_queued` | RunState | Slices waiting for workers | | `slices_completed` | RunState | Successfully processed | @@ -288,7 +264,6 @@ Run 101993 Summary: └── Agents involved: ├── daq_simulator-agent-wenauseic-484 ├── data-agent-wenauseic-481 - ├── fastmon-agent-wenauseic-482 └── fast_processing-agent-wenauseic-483 ``` @@ -382,8 +357,8 @@ chmod +x run_fast_processing.sh - Logs "Fast Processing Agent ready" to console 3. **During Run**: - - Processes `tf_file_registered` messages - - Creates TF slices (15 per TF sample by default) + - Processes `stf_ready` messages + - Creates TF slices (15 per STF by default) - Sends slices to `/topic/panda.slices` - Broadcasts `run_imminent` and `end_run` to `/topic/panda.workers` - Receives and processes slice results @@ -396,8 +371,8 @@ Watch the console output for: [INFO] Fast Processing Agent ready [INFO] Received run_imminent for run 101993 [INFO] Broadcasting run_imminent to workers (target: 30) -[INFO] Received tf_file_registered: swf.101993.000001_tf_001.tf -[INFO] Created 15 slices for TF swf.101993.000001_tf_001.tf +[INFO] Received stf_ready: swf.101993.000001.stf +[INFO] Created 15 slices for STF swf.101993.000001.stf [INFO] Sent slice swf.101993.000001_slice_000.tf to queue [INFO] Received slice_result: slice_000 completed (success) ``` diff --git a/docs/images/fast-processing-pipeline-v11.svg b/docs/images/fast-processing-pipeline-v11.svg new file mode 100644 index 0000000..8f0a091 --- /dev/null +++ b/docs/images/fast-processing-pipeline-v11.svg @@ -0,0 +1,141 @@ + + + + + + + + + + + + + + + + + + + + + + DAQ Simulator + + + + stf_gen + + + + STF + + + + + + + Data Agent + + + + stf_ready + + + + Fast Processing Agent + + + + + + + TF Slices + + + slice 1 + + slice 2 + + slice 3 + + ... + + + + + + + + + + PanDA Workers + + + Worker 1 + EICrecon + + Worker 2 + EICrecon + + Worker 3 + EICrecon + + ... + + + + + + + + + + Reconstruction Output + + + + + + + Analytics + + + + Testbed + DB + + + + STF record + + + + slice bookkeeping + + + + + + + iDDS + Workflow Mgmt + + + + PanDA + Workload Mgmt + + + + + + /topic/panda.workers + run_imminent + target_worker_count + + + + + + + + diff --git a/docs/prompt-processing-workflow.md b/docs/prompt-processing-workflow.md index 5c58a8c..a2162d3 100644 --- a/docs/prompt-processing-workflow.md +++ b/docs/prompt-processing-workflow.md @@ -110,6 +110,40 @@ The workflow resolves the STF output directory in this order: 3. `SWF_PROMPT_PROCESSING_CONTAINER` 4. `/tmp` +### Background execution and PanDA submission + +The processing agent should enqueue blocking work with +`BaseAgent.run_in_background` so PanDA, REST, or storage delays do not block the +STOMP receiver and its heartbeats. Background execution does not imply that the +existing handler is safe to run concurrently. + +For the current prompt-processing agent, start with: + +```bash +export SWF_AGENT_MAX_WORKERS=1 +``` + +This preserves serial task submission while freeing the receiver thread. It is +the safe setting while the agent has per-run values in shared instance state +and uses the legacy `PrunScript.main` path, whose sandbox construction and +process-global state are not thread-safe. + +If the worker count is later increased: + +- pass `run_id`, site, dataset names, and output names into the background doer + and keep them local rather than updating shared `self.run_id`, `self.inDS`, or + `self.outDS`; +- retain a semantic dedup key such as `submit::` to suppress + duplicate delivery, while remembering that different keys still run in + parallel; +- protect the complete legacy prun preparation/submission section with one + agent-owned lock acquired inside the doer, not inside `on_message`; +- use unique temporary paths and do not concurrently change the working + directory or process environment. + +The canonical API contract and locking example are in the +[`swf-common-lib` background-execution documentation](https://github.com/BNLNPPS/swf-common-lib#background-execution-baseagentrun_in_background). + ## Running the Workflow ### Prerequisites diff --git a/episodes/__init__.py b/episodes/__init__.py new file mode 100644 index 0000000..81df8f5 --- /dev/null +++ b/episodes/__init__.py @@ -0,0 +1,12 @@ +"""Episode definitions for testbed workflows. + +Each module defines one workflow's episode: which bus messages are +events, how participants are recognized, and the completion pass that +joins late records (docs/agentic-workflow-view.md). The episode +builder agent arms every definition listed here. +""" + +from .prompt_processing import PromptProcessingEpisodes +from .stf_datataking import StfDatatakingEpisodes + +ALL_DEFINITIONS = [PromptProcessingEpisodes, StfDatatakingEpisodes] diff --git a/episodes/backfill.py b/episodes/backfill.py new file mode 100644 index 0000000..0d991c2 --- /dev/null +++ b/episodes/backfill.py @@ -0,0 +1,83 @@ +"""Backfill workflow episodes from recorded messages. + +Replays a past execution's recorded bus messages through the same +definitions and engine the live builder runs, then drives the +completion pass to closure. The episode carries the recorded times, +not the replay time. Bounded by message-log retention. + +Usage (testbed venv, SWF_MONITOR_HTTP_URL and SWF_API_TOKEN in the +environment): + + python -m episodes.backfill [ ...] +""" + +import os +import sys +import time + +import requests + +from swf_common_lib.episodes import EpisodeBuilder, MonitorEpisodeIngest + +from . import ALL_DEFINITIONS + +COMPLETION_ATTEMPTS = 60 +COMPLETION_INTERVAL_SECONDS = 5 + + +def backfill(execution_id: str) -> bool: + base = (os.environ.get('SWF_MONITOR_HTTP_URL') or '').rstrip('/') + token = os.environ.get('SWF_API_TOKEN') or '' + if not base: + print('SWF_MONITOR_HTTP_URL is not set') + return False + + session = requests.Session() + if token: + session.headers['Authorization'] = f'Token {token}' + response = session.get(f'{base}/api/workflow-messages/', + params={'execution_id': execution_id}, + timeout=60) + response.raise_for_status() + rows = response.json() + if isinstance(rows, dict): + rows = rows.get('results') or [] + rows.sort(key=lambda row: row.get('sent_at') or '') + messages = [] + for row in rows: + content = row.get('message_content') or {} + if content.get('execution_id') != execution_id: + continue + # The recorded sent time backs any message whose writer + # stamped no timestamp of its own. + if row.get('sent_at'): + content.setdefault('sent_at', row['sent_at']) + messages.append(content) + if not messages: + print(f'{execution_id}: no recorded messages') + return False + + ingest = MonitorEpisodeIngest(base_url=base, token=token, + builder_identity='episode-backfill') + builder = EpisodeBuilder( + [definition() for definition in ALL_DEFINITIONS], ingest) + handled = sum(1 for m in messages if builder.handle_message(m)) + print(f'{execution_id}: {handled}/{len(messages)} recorded messages ' + f'replayed') + + for _ in range(COMPLETION_ATTEMPTS): + builder.tick() + if not builder.active: + print(f'{execution_id}: episode closed') + return True + time.sleep(COMPLETION_INTERVAL_SECONDS) + print(f'{execution_id}: completion did not converge') + return False + + +if __name__ == '__main__': + if len(sys.argv) < 2: + print(__doc__) + sys.exit(2) + results = [backfill(execution_id) for execution_id in sys.argv[1:]] + sys.exit(0 if all(results) else 1) diff --git a/episodes/common.py b/episodes/common.py new file mode 100644 index 0000000..ba71bea --- /dev/null +++ b/episodes/common.py @@ -0,0 +1,82 @@ +"""Shared message-to-event mapping for testbed workflow episodes. + +Testbed agents stamp every bus message with sender, msg_type, +timestamp, namespace, execution_id, and run_id; the mapping here turns +that shape into episode events and participant sightings without any +workflow-specific knowledge. Workflow definitions subclass +``TestbedEpisodeDefinition`` and add their completion passes. +""" + +from datetime import datetime +from typing import Dict, List, Optional +from zoneinfo import ZoneInfo + +from swf_common_lib.episodes import EpisodeDefinition, utc_now_iso + +#: Payload keys carried into event payloads when present. +PAYLOAD_KEYS = ('filename', 'sequence', 'site', 'req_id', 'input_dataset', + 'dataset', 'state', 'substate', 'reason', 'container') + +#: Testbed agents stamp bus messages with naive local time; the agents +#: run in this zone. +AGENT_ZONE = ZoneInfo('America/New_York') + + +def agent_kind(sender: str) -> str: + """Sender agent name -> participant kind (e.g. 'daq_simulator').""" + return sender.rsplit('-agent-', 1)[0] if '-agent-' in sender else 'agent' + + +def message_time(message: Dict) -> Optional[str]: + """A message's timestamp as timezone-aware ISO, or None. + + Bus timestamps are naive local stamps from the sending agent; the + store accepts only aware times, so the agents' zone is attached. + Writers that stamp nothing (a known gap) fall back to the recorded + sent time (backfill) or the builder's arrival stamp (live). + """ + 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 + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=AGENT_ZONE) + return parsed.isoformat() + return None + + +class TestbedEpisodeDefinition(EpisodeDefinition): + """Message mapping shared by all testbed workflow definitions.""" + + scope = 'testbed' + + def started_at(self, message: Dict) -> str: + return message_time(message) or utc_now_iso() + + def ended_at(self, message: Dict) -> str: + return message_time(message) or utc_now_iso() + + def event_from_message(self, message: Dict) -> Optional[Dict]: + sender = message.get('sender') or message.get('sender_agent') + msg_type = message.get('msg_type') + when = message_time(message) + if not (sender and msg_type and when): + return None + payload = {key: message[key] for key in PAYLOAD_KEYS + if message.get(key) is not None} + return {'time': when, 'kind': msg_type, 'participant': sender, + 'payload': payload} + + def participants_from_message(self, message: Dict) -> List[Dict]: + sender = message.get('sender') or message.get('sender_agent') + when = message_time(message) + if not sender: + return [] + entry = {'id': sender, 'label': sender, 'kind': agent_kind(sender)} + if when: + entry['born_at'] = when + return [entry] diff --git a/episodes/prompt_processing.py b/episodes/prompt_processing.py new file mode 100644 index 0000000..9d76187 --- /dev/null +++ b/episodes/prompt_processing.py @@ -0,0 +1,109 @@ +"""Prompt-processing workflow episodes. + +Agent messages are recorded live through the shared testbed mapping; +the completion pass joins the PanDA side — the tasks named by the run +number and the jobs born within them, which are the workflow's worker +lanes (docs/agentic-workflow-view.md). +""" + +import logging + +from .common import TestbedEpisodeDefinition + +logger = logging.getLogger(__name__) + +TERMINAL_TASK_STATUSES = {'done', 'finished', 'failed', 'broken', + 'aborted', 'exhausted'} + + +def _aware(value): + """PanDA REST timestamps are naive UTC; stamp the offset.""" + if not value: + return None + value = str(value) + if value.endswith('Z') or '+' in value[10:]: + return value + return value + '+00:00' + + +class PromptProcessingEpisodes(TestbedEpisodeDefinition): + workflow_name = 'prompt_processing' + completion_deadline_seconds = 1800 + + def completion_poll(self, context, ingest): + run_id = ((context.last_message or {}).get('run_id') + or (context.first_message or {}).get('run_id')) + if not run_id: + logger.warning('episode %s has no run id; closing without ' + 'a PanDA join', context.episode_id) + return True + + session = ingest.session + base = ingest.base_url + response = session.get( + f'{base}/api/panda/tasks/', + params={'taskname': f'swf.{run_id}.processed', 'days': 2}, + timeout=30) + response.raise_for_status() + tasks = response.json().get('items') or [] + if not tasks: + # Tasks appear seconds after submission; none yet means the + # join is early, not empty. The deadline bounds the wait. + return False + if any((t.get('status') or '') not in TERMINAL_TASK_STATUSES + for t in tasks): + return False + + events, participants = [], [] + for task in tasks: + taskid = task['jeditaskid'] + task_pid = f'task-{taskid}' + participants.append({ + 'id': task_pid, + 'label': task.get('taskname') or task_pid, + 'kind': 'panda_task', + 'born_at': _aware(task.get('creationdate')), + 'died_at': _aware(task.get('endtime')), + }) + events.append({ + 'time': _aware(task.get('creationdate')), + 'kind': 'task_created', + 'participant': task_pid, + 'payload': {'jeditaskid': taskid, + 'site': task.get('site'), + 'status': task.get('status')}, + }) + jobs_response = session.get( + f'{base}/api/panda/jobs/', + params={'taskid': taskid, 'days': 2}, timeout=60) + jobs_response.raise_for_status() + for job in jobs_response.json().get('items') or []: + job_pid = f"job-{job['pandaid']}" + participants.append({ + 'id': job_pid, + 'label': f"job {job['pandaid']}", + 'kind': 'panda_job', + 'born_at': _aware(job.get('creationtime')), + 'died_at': _aware(job.get('endtime')), + }) + for field, kind in (('creationtime', 'job_created'), + ('starttime', 'job_started'), + ('endtime', 'job_ended')): + if job.get(field): + events.append({ + 'time': _aware(job[field]), + 'kind': kind, + 'participant': job_pid, + 'payload': {'site': job.get('computingsite'), + 'status': job.get('jobstatus'), + 'jeditaskid': taskid}, + }) + + ingest.append(scope=self.scope, episode_id=context.episode_id, + events=events, participants=participants) + context.notes['panda_tasks'] = [t['jeditaskid'] for t in tasks] + return True + + def summary(self, context): + return {'run_id': ((context.last_message or {}).get('run_id')), + 'panda_tasks': context.notes.get('panda_tasks', [])} diff --git a/episodes/stf_datataking.py b/episodes/stf_datataking.py new file mode 100644 index 0000000..d325e01 --- /dev/null +++ b/episodes/stf_datataking.py @@ -0,0 +1,16 @@ +"""STF datataking workflow episodes. + +The default testbed exercise: agent messages recorded live through the +shared testbed mapping, no workload join — the workflow's example +agents submit nothing. +""" + +from .common import TestbedEpisodeDefinition + + +class StfDatatakingEpisodes(TestbedEpisodeDefinition): + workflow_name = 'stf_datataking' + completion_deadline_seconds = 60 + + def summary(self, context): + return {'run_id': (context.last_message or {}).get('run_id')} diff --git a/example_agents/fast_processing_agent.py b/example_agents/fast_processing_agent.py index d83cb81..4fb2298 100644 --- a/example_agents/fast_processing_agent.py +++ b/example_agents/fast_processing_agent.py @@ -683,9 +683,14 @@ def _create_tf_slices(self, tf_filename, stf_filename, tf_first, tf_last, tf_cou } } - # Create in database + # Create in database. The record references its registered + # TF sample (FastMonFile) by name; the derived slice filename + # stays in the queue message only. + db_payload = {k: v for k, v in slice_data.items() + if k not in ('tf_filename', 'stf_filename')} + db_payload['fastmon_file'] = tf_filename try: - result = self.call_monitor_api('POST', '/tf-slices/', slice_data) + result = self.call_monitor_api('POST', '/tf-slices/', db_payload) if result: self.stats['slices_created'] += 1 self.slices_created += 1 diff --git a/src/swf_testbed_cli/main.py b/src/swf_testbed_cli/main.py index 9b0598a..20cda18 100644 --- a/src/swf_testbed_cli/main.py +++ b/src/swf_testbed_cli/main.py @@ -1,6 +1,7 @@ import typer import shutil from pathlib import Path +import socket import subprocess import os import sys @@ -105,43 +106,36 @@ def _check_supervisord_running() -> bool: except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError): return False -def _check_postgres_connection(): - """Checks the connection to the PostgreSQL database.""" - db_host = os.getenv("DB_HOST", "localhost") - db_port = os.getenv("DB_PORT", "5432") - db_user = os.getenv("DB_USER", "admin") - db_name = os.getenv("DB_NAME", "swfdb") - - print(f"--- Checking PostgreSQL connection at {db_host}:{db_port} ---") +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 +def _check_postgres_connection(): + """Checks that PostgreSQL is accepting connections.""" + return _check_tcp_service( + "PostgreSQL", + os.getenv("DB_HOST", "localhost"), + os.getenv("DB_PORT", "5432"), + ) + def _check_activemq_connection(): - """Checks if ActiveMQ is listening on its port.""" - amq_port = os.getenv("ACTIVEMQ_PORT", "61616") - print(f"--- Checking ActiveMQ connection on port {amq_port} ---") - result = subprocess.run(f"lsof -i -P -n | grep LISTEN | grep ':{amq_port}'", shell=True, capture_output=True) - if result.returncode == 0: - print(f"ActiveMQ appears to be running and listening on port {amq_port}.") - return True - else: - print(f"Warning: Could not detect a service listening on port {amq_port}.") - print("Please ensure ActiveMQ is running.") - return False + """Checks that ActiveMQ is accepting connections.""" + return _check_tcp_service( + "ActiveMQ", + os.getenv("ACTIVEMQ_HOST", "localhost"), + os.getenv("ACTIVEMQ_PORT", "61616"), + ) def _get_workflow_status(): @@ -331,10 +325,19 @@ def stop_agents(): print(f"Error: {AGENTS_CONF} not found. Run 'testbed init' first.") raise typer.Exit(code=1) - # Stop all agents + # 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 + print("Stopping workflow agents...") + to_stop = get_running_agents() + if not to_stop: + print("No workflow agents running.") + return result = subprocess.run( - ["supervisorctl", "-c", str(conf_path), "stop", "all"], + ["supervisorctl", "-c", str(conf_path), "stop"] + to_stop, capture_output=True, text=True, cwd=testbed_root diff --git a/workflows/fast_processing_default.toml b/workflows/fast_processing_default.toml index 18f60bf..d483429 100644 --- a/workflows/fast_processing_default.toml +++ b/workflows/fast_processing_default.toml @@ -8,7 +8,7 @@ # Usage: testbed run fast_processing_default [testbed] -namespace = "torre2" +namespace = "torre1" [workflow] name = "fast_processing" diff --git a/workflows/orchestrator.py b/workflows/orchestrator.py index 2511488..f3d10e3 100644 --- a/workflows/orchestrator.py +++ b/workflows/orchestrator.py @@ -26,6 +26,13 @@ } AGENTS_CONF = 'agents.supervisord.conf' + +# Standing infrastructure programs: not part of any workflow's agent +# lifecycle, so run/stop checks leave them alone. Empty since the +# episode builder moved to the swf-episode-builder systemd service; +# the mechanism stays for future supervisord-resident standing +# programs. +STANDING_PROGRAMS = set() AGENTS_SOCK = '/tmp/swf-agents-supervisor.sock' @@ -88,7 +95,22 @@ def restart_supervisord() -> bool: capture_output=True, cwd=testbed_dir ) - time.sleep(1) + # Shutdown stops every program, including standing ones such + # as the episode builder, and can take several seconds; the + # new instance cannot bind until the old one is gone. + for _ in range(30): + time.sleep(1) + probe = subprocess.run( + ['supervisorctl', '-c', conf_path, 'status'], + capture_output=True, + text=True, + cwd=testbed_dir + ) + if probe.returncode == 4: + break + else: + print("Warning: previous supervisord did not shut down " + "within 30 s") # Start fresh print("Starting supervisord...") @@ -185,7 +207,8 @@ def get_running_agents() -> list: if 'RUNNING' in line: # Line format: "program-name RUNNING pid 12345, uptime 0:00:05" program_name = line.split()[0] - running.append(program_name) + if program_name not in STANDING_PROGRAMS: + running.append(program_name) return running diff --git a/workflows/testbed.toml b/workflows/testbed.toml index 19c6876..eb560fe 100644 --- a/workflows/testbed.toml +++ b/workflows/testbed.toml @@ -30,7 +30,10 @@ namespace = "torre1" # Scripts are relative to swf-testbed root. [agents.data] -enabled = false +# The data agent is the epictopic listener: it registers runs and STF +# files in the monitor and feeds the processing queue. The processing +# agent alone hears nothing (queue-based messaging since Feb 2026). +enabled = true script = "example_agents/example_data_agent.py" [agents.processing] diff --git a/workflows/workflow_runner.py b/workflows/workflow_runner.py index 4e92ec6..f85d3a1 100644 --- a/workflows/workflow_runner.py +++ b/workflows/workflow_runner.py @@ -310,7 +310,7 @@ def run_workflow(self, workflow_name: str, config_name: Optional[str] = None, # workflow must not leave its announced runs claiming activity. self._announced_runs = set() try: - self._execute_workflow( + outcome = self._execute_workflow( execution_id=execution_id, workflow_code=workflow_code, config=config, @@ -321,7 +321,14 @@ def run_workflow(self, workflow_name: str, config_name: Optional[str] = None, self._update_execution_status(execution_id, 'failed') self._abandon_announced_runs() raise - self._update_execution_status(execution_id, 'completed') + if outcome == 'stopped': + # A stop mid-run is not a completion: the workflow never sent + # end_run, so its announced runs must not be left claiming + # activity. + self._update_execution_status(execution_id, 'terminated') + self._abandon_announced_runs() + else: + self._update_execution_status(execution_id, 'completed') return execution_id @@ -631,14 +638,16 @@ def _execute_workflow(self, execution_id: str, workflow_code: str, end_time = duration if duration and duration > 0 else float('inf') while True: - # Step callback - check stop flag and other per-step actions - if not self._on_simulation_step(env, execution_id): - break - - # Check if workflow process completed + # Check if workflow process completed — before the stop + # check, so a finished workflow always reads 'completed' + # even when a stop request races its last event. if workflow_process.processed: break + # Step callback - check stop flag and other per-step actions + if not self._on_simulation_step(env, execution_id): + return 'stopped' + # Check duration limit if env.now >= end_time: self.logger.info(f"Duration limit reached: {duration}s") @@ -652,6 +661,7 @@ def _execute_workflow(self, execution_id: str, workflow_code: str, break else: raise ValueError("WorkflowExecutor class not found in workflow code") + return 'completed' def _abandon_announced_runs(self): """Terminalize the runs this execution announced but did not