Publish stf_processed on end of processing - #71
Conversation
There was a problem hiding this comment.
Pull request overview
This PR updates the prompt processing agent to publish an stf_processed ActiveMQ message when PanDA polling completes (or times out), and propagates output dataset context through the polling lifecycle so downstream consumers can associate results with the produced dataset.
Changes:
- Convert PanDA job IDs to integers before calling
getFullJobStatus(). - Add
output_datasetpropagation to the PanDA polling scheduler task context and recovery path. - Publish an
stf_processedmessage to/topic/epictopicwhen a polled PanDA task reaches a terminal condition.
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
|
I have an overall question about using ActiveMQ. For the downstream consumers, won't it be able to use the |
| ) | ||
| ) | ||
| # Publish stf_processed message to ActiveMQ | ||
| self._publish_stf_processed(task, result, timed_out) |
There was a problem hiding this comment.
result from last poll may not provide the expected numbers, because each poll initializes at line 889, the for last poll processed or failed is 0 if earlier polls have already patched all STF files. The processed/failed in each poll is not the final statistics as reported to swf-monitor.
This is the tail of run 102966 polls:
2026-08-17 15:58:34,205 - INFO - base_agent - PanDA polling updated STF files for run 102966 (task_id=38974, site=E1_BNL, input_dataset=group.daq:run.102966.E1_BNL): processed=1, failed=0, task_status=running, jobs_seen=8, unfinished=0, unmatched=0
INFO: PanDA polling updated STF files for run 102966 (task_id=38975, site=E1_JLAB, input_dataset=group.daq:run.102966.E1_JLAB): processed=1, failed=0, task_status=running, jobs_seen=7, unfinished=0, unmatched=0
2026-08-17 15:58:54,466 - INFO - base_agent - PanDA polling updated STF files for run 102966 (task_id=38975, site=E1_JLAB, input_dataset=group.daq:run.102966.E1_JLAB): processed=1, failed=0, task_status=running, jobs_seen=7, unfinished=0, unmatched=0
INFO: PanDA polling updated STF files for run 102966 (task_id=38974, site=E1_BNL, input_dataset=group.daq:run.102966.E1_BNL): processed=0, failed=0, task_status=done, jobs_seen=8, unfinished=0, unmatched=0
2026-08-17 15:59:34,944 - INFO - base_agent - PanDA polling updated STF files for run 102966 (task_id=38974, site=E1_BNL, input_dataset=group.daq:run.102966.E1_BNL): processed=0, failed=0, task_status=done, jobs_seen=8, unfinished=0, unmatched=0
INFO: PanDA polling updated STF files for run 102966 (task_id=38975, site=E1_JLAB, input_dataset=group.daq:run.102966.E1_JLAB): processed=0, failed=0, task_status=done, jobs_seen=7, unfinished=0, unmatched=0
2026-08-17 15:59:55,655 - INFO - base_agent - PanDA polling updated STF files for run 102966 (task_id=38975, site=E1_JLAB, input_dataset=group.daq:run.102966.E1_JLAB): processed=0, failed=0, task_status=done, jobs_seen=7, unfinished=0, unmatched=0
There was a problem hiding this comment.
Maybe downstream can aggregate the processed/failed itself or using swf-monitor API, or we can make processed/failed less ambiguous in the message, e.g. updated_processed/updated_failed?
There was a problem hiding this comment.
I don't see what is the issue. This is triggered when task enters completed state. You had two tasks with 8 and 7 jobs (15 together), this should have reported two messages.
There was a problem hiding this comment.
Depending on how the message will be used, I just want to pointed out the message from each poll only gives processed/failed count in that poll, not total and last poll even gives 0. If the downstream only needs task status, that's fine.
There was a problem hiding this comment.
I think the consumer will need just the status and dataset name. In my tests I got the correct final values, but if you reproduce something else, we can fix that.
By that logic we should excise ActiveMQ everywhere. We'd need to first port data-agent and processing-agent to use monitor API polling. |
| 'run_id': task.get("run_number"), | ||
| 'panda_task_id': task.get("panda_task_id"), | ||
| 'task_status': result.get("task_status"), | ||
| 'processed': result.get("processed", 0), |
There was a problem hiding this comment.
Here the key might be misleading if one thinks processed is the total. It's actually my fault to name it that way. Its true meaning is updated_processed in each poll.
There was a problem hiding this comment.
So it's a difference, not an absolute value?
There was a problem hiding this comment.
We could remove this field and add it back once correct accounting is implemented.
There was a problem hiding this comment.
Remove them are also fine if the meaningful field is just task_status. I don't think we should aggregate the total processed/failed in each poll though, otherwise the statistics will be wrong in the monitor. Each poll only tries to update previously unfinished STF status. If need the total statistics, I think stf-files in the monitor is a better method to extract that.
zhaoyuyoung
left a comment
zhaoyuyoung
left a comment
There was a problem hiding this comment.
The added stf_processed is good in terms of codes and as long as the message items are clear for downstream consumers, it's fine to merge.
|
Hi @veprbl, I approved this PR. I also tested it in a prompt processing workflow and saw the new |
No description provided.