Added the decision box for prompt processing workflow

agents.supervisord.conf

@@ -85,7 +85,7 @@ stderr_logfile=%(here)s/logs/%(program_name)s.log
 [program:stf-data-agent]
 command=python -u agents/data_agent.py -v
 directory=%(ENV_SWF_HOME)s/swf-testbed
-environment=SWF_TESTBED_CONFIG="%(ENV_SWF_TESTBED_CONFIG)s"
+environment=SWF_TESTBED_CONFIG="%(ENV_SWF_TESTBED_CONFIG)s",SWF_AGENT_MAX_WORKERS="1"
 autostart=false
 autorestart=true
 stopwaitsecs=10
@@ -96,7 +96,7 @@ stderr_logfile=%(here)s/logs/%(program_name)s.log
 [program:stf-processing-agent]
 command=python -u agents/prompt_processing_agent.py -v
 directory=%(ENV_SWF_HOME)s/swf-testbed
-environment=SWF_TESTBED_CONFIG="%(ENV_SWF_TESTBED_CONFIG)s"
+environment=SWF_TESTBED_CONFIG="%(ENV_SWF_TESTBED_CONFIG)s",SWF_AGENT_MAX_WORKERS="1"
 autostart=false
 autorestart=true
 stopwaitsecs=10

docs/decision-box.md

@@ -0,0 +1,123 @@
+# ePIC Decision Box
+
+`swf_testbed_decision_box` implements the site-specific dataset control pattern
+for prompt processing:
+
+- one full open Rucio dataset per run, containing all STF file DIDs
+- one open processing dataset per site, such as `group.daq:run.123.E1_BNL`
+- the prompt-processing decision box decides which site dataset(s) receive each file DID
+- PanDA/JEDI consumes the site-specific datasets with `runUntilClosed=True`
+
+The implementation is part of `swf-testbed` and uses Rucio for dataset
+creation, file attachment, and dataset closure.
+
+## Flow
+
+![Prompt processing workflow with decision box](images/decision-box-flow.svg)
+
+In the diagram, `Data agent -> Processing agent` is the `stf_ready`
+notification. The dotted arrows from the full run dataset to the BNL/JLAB
+subset datasets mean the subset datasets contain the same STF file DIDs selected
+from the full dataset.
+
+## Dataset Ownership
+
+The data agent owns the full run dataset:
+
+```text
+group.daq:swf.<run>.run
+```
+
+It creates the full dataset on `run_imminent` and attaches every STF file DID
+to it on `stf_gen`.
+
+The data agent also owns the site-specific processing datasets used by the
+decision box:
+
+```text
+group.daq:run.<run>.E1_BNL
+group.daq:run.<run>.E1_JLAB
+```
+
+It creates those datasets on `run_imminent`, after creating the full run
+dataset. They are logical subsets of the full run dataset: the decision box
+attaches the same STF file DID, not a PFN copy, to only the site dataset(s)
+selected by policy. The data agent records the decision metadata in
+swf-monitor for the processing agent to consume.
+
+If `decision_box_site_dataset_template` is unset, the package default is to
+derive site datasets from the full run dataset, for example
+`group.daq:swf.<run>.run.E1_BNL`.
+
+## Prompt Processing Integration
+
+`swf-testbed/agents/data_agent.py` uses this package when
+`[prompt_processing].decision_box_enabled = true`.
+
+In that mode the data agent:
+
+- creates the site-specific processing datasets
+- applies decisions for each `stf_gen` message after attaching the STF DID to
+  the full run dataset
+- attaches the same STF DID to the selected site-specific datasets
+- records the decision in the STF row metadata
+- sends one site-specific `stf_ready` message the first time a site-specific
+  dataset receives an STF DID
+- closes the site-specific processing datasets on `end_run`
+
+`swf-testbed/agents/prompt_processing_agent.py` does not mutate those input
+datasets. It submits a `runUntilClosed=True` PanDA task only for the site named
+in each `stf_ready` message, then uses the data-agent decision metadata to
+claim and poll STF processing status. If no STF is selected for a site during a
+run, no empty task is submitted for that site.
+
+The prompt-processing workflow config enables the Rucio-backed decision box:
+
+```toml
+decision_box_enabled = true
+decision_box_policy = "round-robin"
+decision_box_sites = ["E1_BNL", "E1_JLAB"]
+decision_box_rucio_scope = "group.daq"
+decision_box_site_dataset_template = "run.{run_number}.{site_name}"
+```
+
+With this template, a full run dataset such as `group.daq:swf.102741.run`
+produces site-specific processing datasets such as
+`group.daq:run.102741.E1_BNL` and `group.daq:run.102741.E1_JLAB`. These logical
+work-partition datasets do not match broad `group.daq:swf*` Rucio rules.
+
+The decision box expects the same Rucio client and `rucio_comms` environment
+used by the existing data agent.
+
+When the decision box is disabled, prompt processing falls back to one PanDA
+task over the full run dataset. That legacy task uses
+`non_decision_box_site`, which defaults to `E1_BNL` and can be overridden by
+`SWF_NON_DECISION_BOX_SITE`.
+
+## Policy Modes
+
+- `round-robin`: alternate assignments across the configured sites
+- `hash`: deterministic assignment based on the file DID
+- `both`: assign each file to all configured sites
+- `none`: do not attach the file to any site-specific processing dataset
+- `explicit`: use the sites supplied in the incoming message fields
+
+Policies implement `DecisionPolicy._choose_sites(context)` and return a
+`SiteAssignment`. The `DecisionContext` contains the file DID, run dataset,
+run number, configured sites, sequence number, original message fields, stored
+run conditions, and optional policy metadata. This keeps the placeholder
+policies small while giving future experiment policies access to detector state,
+run configuration, operator input, or other decision inputs without changing the
+dataset mutation service.
+
+## PanDA Submission Shape
+
+The processing tasks consume the site datasets, not the full run dataset:
+
+```text
+--site E1_BNL  --inDS group.daq:run.101871.E1_BNL  runUntilClosed=True
+--site E1_JLAB --inDS group.daq:run.101871.E1_JLAB runUntilClosed=True
+```
+
+Use split settings such as `nFilesPerJob=1` and `nChunksToWait=1` when files
+should be released promptly.

src/swf_agent_lib/__init__.py

@@ -0,0 +1,2 @@
+"""Shared helpers for SWF testbed agents."""
+

src/swf_agent_lib/config_helpers.py

@@ -0,0 +1,127 @@
+import os
+import tomllib
+
+
+class PromptProcessingConfigMixin:
+    """Shared prompt-processing config helpers for local testbed agents."""
+
+    def _prompt_processing_config_path(self):
+        return os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "workflows", "prompt_processing.toml")
+
+    def _load_prompt_processing_section(self, config_path, warn=False):
+        if not config_path:
+            return {}
+        try:
+            with open(config_path, "rb") as config_file:
+                return tomllib.load(config_file).get("prompt_processing", {})
+        except (OSError, TypeError, tomllib.TOMLDecodeError) as e:
+            if warn:
+                self.logger.warning(
+                    f"Could not load prompt_processing config from {config_path}: {e}",
+                    extra=self._log_extra()
+                )
+            return {}
+
+    def _load_prompt_processing_config(self):
+        """Load prompt-processing settings, with workflow defaults plus active config overrides."""
+        prompt_config = self._load_prompt_processing_section(self._prompt_processing_config_path(), warn=True)
+        active_config = self._load_prompt_processing_section(self.config_path, warn=True)
+        prompt_config.update(active_config)
+        return prompt_config
+
+    def _config_bool(self, config, key, env_var, default):
+        """Read a boolean setting from config, with an environment override."""
+        value = os.getenv(env_var, config.get(key, default))
+        if isinstance(value, bool):
+            return value
+        if isinstance(value, str):
+            return value.strip().lower() in {"1", "true", "yes", "on"}
+        return bool(value)
+
+    def _config_int(self, config, key, env_var, default):
+        """Read an integer setting from config, with an environment override."""
+        value = os.getenv(env_var, config.get(key, default))
+        try:
+            return int(value)
+        except (TypeError, ValueError):
+            self.logger.warning(
+                f"Invalid {key} value {value!r}; using default {default}",
+                extra=self._log_extra()
+            )
+            return default
+
+    def _config_list(self, config, key, env_var, default):
+        """Read a comma-separated list setting from config, with an environment override."""
+        value = os.getenv(env_var, config.get(key, default))
+        if isinstance(value, str):
+            return [item.strip() for item in value.split(",") if item.strip()]
+        if isinstance(value, (list, tuple)):
+            return [str(item).strip() for item in value if str(item).strip()]
+        return list(default)
+
+    def _message_bool(self, message_data, key, default):
+        value = message_data.get(key, default)
+        if isinstance(value, bool):
+            return value
+        if isinstance(value, str):
+            return value.strip().lower() in {"1", "true", "yes", "on"}
+        return bool(value)
+
+
+class DecisionDatasetNamingMixin:
+    """Shared decision-box message and dataset helpers."""
+
+    def _decision_box_context_for_run(self, run_id):
+        return {}
+
+    def _decision_box_enabled_for_message(self, message_data, run_id=None):
+        if "decision_box_enabled" in message_data:
+            return self._message_bool(message_data, "decision_box_enabled", self.decision_box_enabled)
+        if run_id is not None:
+            context = self._decision_box_context_for_run(run_id)
+            if "decision_box_enabled" in context:
+                return bool(context["decision_box_enabled"])
+        return self.decision_box_enabled
+
+    def _non_decision_box_site_for_message(self, message_data, run_id=None):
+        site = message_data.get("non_decision_box_site")
+        if site:
+            return str(site).strip()
+        if run_id is not None:
+            context = self._decision_box_context_for_run(run_id)
+            site = context.get("non_decision_box_site")
+            if site:
+                return str(site).strip()
+        return getattr(self, "non_decision_box_site", None)
+
+    def _run_dataset_name(self, run_number=None):
+        dataset = getattr(self, "dataset", None)
+        if dataset:
+            return dataset
+        if run_number is not None:
+            return f"swf.{run_number}.run"
+        return ""
+
+    def _run_dataset_did(self, run_number=None):
+        return f"{self.decision_box_rucio_scope}:{self._run_dataset_name(run_number)}"
+
+    def _input_dataset_name_for_site(self, run_number, site_name):
+        run_dataset_name = f"swf.{run_number}.run"
+        if self.decision_box_site_dataset_template:
+            return self.decision_box_site_dataset_template.format(
+                run_dataset_name=run_dataset_name,
+                run_number=run_number,
+                site_name=site_name,
+                site=site_name,
+            )
+        return f"{self._run_dataset_name(run_number)}.{site_name}"
+
+    def _input_dataset_did_for_site(self, run_number, site_name):
+        return f"{self.decision_box_rucio_scope}:{self._input_dataset_name_for_site(run_number, site_name)}"
+
+    def _site_name_for_dataset(self, run_number, dataset_did):
+        for site_name in self.decision_box_sites:
+            if dataset_did == self._input_dataset_did_for_site(run_number, site_name):
+                return site_name
+        return None
+

src/swf_testbed_decision_box/__init__.py

@@ -0,0 +1,18 @@
+"""Decision box prototype for site-specific STF processing datasets."""
+
+from .models import Decision, DecisionContext, FileDID, Site, SiteAssignment
+from .monitor_metadata import execution_id_matches, metadata_with_execution_id
+from .policy import build_policy
+from .service import DecisionBox
+
+__all__ = [
+    "Decision",
+    "DecisionContext",
+    "DecisionBox",
+    "FileDID",
+    "Site",
+    "SiteAssignment",
+    "build_policy",
+    "execution_id_matches",
+    "metadata_with_execution_id",
+]