swf-testbed/agents/episode_builder_agent.py

0001 #!/usr/bin/env python3
0002 """Episode builder agent: records workflow episodes from bus traffic.
0003
0004 One more listener on the epictopic, in keeping with the open-listening
0005 messaging philosophy: it consumes every message, routes executions to
0006 the armed episode definitions (the episodes package), and drives each
0007 episode through open, append, completion, and close against the
0008 monitor's episode ingest REST (docs/agentic-workflow-view.md).
0009
0010 Episodes are scope-wide, so unlike workflow agents this agent applies
0011 no namespace filtering: every namespace's executions are recorded.
0012 """
0013
0014 import json
0015 import logging
0016 import sys
0017 from pathlib import Path
0018
0019 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
0020
0021 from swf_common_lib.base_agent import BaseAgent
0022 from swf_common_lib.episodes import EpisodeBuilder, MonitorEpisodeIngest
0023
0024 from episodes import ALL_DEFINITIONS
0025
0026
0027 class EpisodeBuilderAgent(BaseAgent):
0028     def __init__(self):
0029         super().__init__(agent_type='EPISODE_BUILDER',
0030                          subscription_queue='/topic/epictopic')
0031         # The builder identity is stable across restarts and instances:
0032         # a restarted builder must resume the episodes its predecessor
0033         # opened. The per-process agent name would strand them.
0034         ingest = MonitorEpisodeIngest(
0035             base_url=self.base_url,
0036             token=self.api_token,
0037             builder_identity='episode-builder',
0038         )
0039         self.builder = EpisodeBuilder(
0040             [definition() for definition in ALL_DEFINITIONS], ingest)
0041         logging.info('armed definitions: %s',
0042                      [d.workflow_name for d in self.builder.definitions])
0043         adopted = self.builder.adopt_open_episodes()
0044         if adopted:
0045             logging.info('adopted %d open episode(s) from a previous '
0046                          'builder instance', adopted)
0047
0048     def on_message(self, frame):
0049         try:
0050             message = json.loads(frame.body)
0051         except (ValueError, TypeError) as exc:
0052             logging.error('unparseable message dropped: %s', exc)
0053             return
0054         self.builder.handle_message(message)
0055
0056     def send_heartbeat(self):
0057         result = super().send_heartbeat()
0058         # The heartbeat cycle is the completion-poll cadence.
0059         self.builder.tick()
0060         return result
0061
0062
0063 if __name__ == '__main__':
0064     EpisodeBuilderAgent().run()