Add aggregate wakeups for grouped remote jobs

docs/remote_job_monitoring.md

@@ -152,7 +152,9 @@ from the current one — see [Provider Plugin Architecture](#provider-plugin-arc
 
 Batch Job `prepared`/`pending` normalize to `queued`, `active`/`running` to
 `running`, `succeeded` to `succeeded`, `failed` to `failed`, and
-`deleted`/`killed` to `cancelled`. `prepared` remains visibly incomplete;
+`deleted`/`killed`/`stopped`/`terminated`/`cancelled` to `cancelled`, with
+`termination_origin=provider_observed`. This records what the provider reported
+without claiming which person or system initiated the stop. `prepared` remains visibly incomplete;
 unknown statuses are nonterminal observations with no guessed transition.
 Use `status_name`, `terminal`, and concrete `errorMessage` / `errorCode` details,
 never numeric backend status or `exitCode`. Curated snapshots must not persist
@@ -201,6 +203,15 @@ terminating its sandbox allocation. Initial terminal probes and explicit
 refreshes use the same transactional path. Queue-to-running updates do not
 invoke the agent.
 
+For parallel work, an agent can call `create_remote_job_group` with an exact
+expected member count, then pass the returned `group_id` to each Batch Job
+submission or attachment. Grouped jobs retain their individual lifecycle
+events but suppress individual lifecycle wakeups. The group creates one durable
+aggregate notification when all expected jobs have outcomes, when its optional
+failed-job ratio is reached, or when its deadline expires (48 hours by default). The first
+policy reached wins, and its notification includes a snapshot of every current
+member. Group policy and delivery state survive control-plane restarts.
+
 The durable outbox claims notifications with leases, defers busy sessions,
 retries delivery errors with bounded backoff, and records delivery status,
 attempts, last error, and the managed run ID. Exhausted delivery retries remain

src/matcreator/agents/execution_agent/remote_job_tools.py

@@ -60,6 +60,7 @@ def _submit(
     spec: dict[str, Any],
     discriminator: str,
     persisted_specification: dict[str, Any] | None = None,
+    group_id: str | None = None,
 ) -> dict[str, Any]:
     """Shared submission plumbing used by every provider-specific submit tool."""
     session_id = str(tool_context.state.get("session_id") or "")
@@ -77,6 +78,7 @@ def _submit(
             idempotency_key=idempotency_key,
             spec=spec,
             persisted_specification=persisted_specification,
+            group_id=group_id,
         )
     except Exception as exc:
         return {"status": "error", "message": f"{provider} submission failed: {exc}"}
@@ -302,9 +304,14 @@ def submit_bohr_batchjob(
     out_files: list[str] | None = None,
     max_run_time: str = "24h",
     max_wait_time: str = "30m",
+    group_id: str | None = None,
 ) -> dict[str, Any]:
     """Submit or reuse a tracked sandbox-based job through `bohr batchjob submit`.
 
+    When this step will submit two or more similar or simultaneous Batch Jobs,
+    first call create_remote_job_group once and pass its returned group_id to
+    every related submission.
+
     Supply exactly one of machine_type or sku_id, discovered with
     `bohr batchjob machine list -o json`. project_id falls back to
     BOHRIUM_PROJECT_ID. input_path is a path RELATIVE to the step's working
@@ -390,6 +397,7 @@ def submit_bohr_batchjob(
         provider="bohr_batchjob",
         spec=spec,
         discriminator=f"bohr_batchjob:{name}",
+        group_id=group_id,
     )
     return _submission_response(
         result,
@@ -405,9 +413,13 @@ def attach_bohr_batchjob(
     tool_context: ToolContext,
     *,
     batchjob_id: str,
+    group_id: str | None = None,
 ) -> dict[str, Any]:
     """Attach an already-submitted Batch Job by its explicit string ID; never submit.
 
+    When attaching two or more related Batch Jobs together, first call
+    create_remote_job_group once and pass its group_id to every attachment.
+
     Uses the existing bohr account authentication to read the remote status.
     Repeated attachment reuses the current session's durable job record.
     Use the returned job_id for status, controls, and output collection.
@@ -430,6 +442,7 @@ def attach_bohr_batchjob(
             external_id=batchjob_id,
             node_id=node_id,
             step_number=tool_context.state.get("step_number"),
+            group_id=group_id,
         )
     except Exception as exc:
         error = f"bohr batchjob attachment failed: {exc}"
@@ -452,10 +465,52 @@ def attach_bohr_batchjob(
     }
 
 
+def create_remote_job_group(
+    tool_context: ToolContext,
+    *,
+    name: str,
+    expected_jobs: int,
+    failure_ratio: float | None = None,
+    deadline_seconds: float = 172800,
+) -> dict[str, Any]:
+    """Create or reuse a durable group that emits one aggregate agent wakeup.
+
+    ALWAYS call this once before submitting or attaching two or more similar or
+    simultaneous Batch Jobs in one step. Set expected_jobs to the exact count.
+    Add the returned group_id to each submit_bohr_batchjob or
+    attach_bohr_batchjob call. The group wakes once when all expected jobs have
+    outcomes, the optional failed-job ratio is reached, or the optional
+    deadline elapses (48 hours by default). Grouped jobs do not create
+    individual lifecycle wakeups.
+    """
+    session_id = str(tool_context.state.get("session_id") or "")
+    if not session_id:
+        return {"status": "error", "message": "No session_id is available for remote-job grouping."}
+    try:
+        group = _service().create_job_group(
+            owner_id=_owner_id(tool_context),
+            session_id=session_id,
+            name=name,
+            expected_jobs=expected_jobs,
+            failure_ratio=failure_ratio,
+            deadline_seconds=deadline_seconds,
+        )
+    except Exception as exc:
+        return {"status": "error", "message": f"Remote job group creation failed: {exc}"}
+    return {
+        "status": "ready",
+        "group_id": group["group_id"],
+        "name": group["name"],
+        "expected_jobs": group["expected_jobs"],
+        "failure_ratio": group["failure_ratio"],
+        "deadline_at": group["deadline_at"],
+    }
+
+
 def list_remote_jobs(tool_context: ToolContext, active_only: bool = False) -> dict[str, Any]:
     """List remote jobs tracked for the current session, newest-updated first.
 
-    Returns a compact per-job projection (``job_id``, ``provider``, ``node_id``,
+    Returns a compact per-job projection (``job_id``, ``group_id``, ``provider``, ``node_id``,
     ``status``, ``external_id``, ``updated_at``, ``error``) instead of full
     snapshots/events, so it is cheap to call before answering questions about
     running jobs or after a restart, without resorting to ``read_session_log``
@@ -469,7 +524,13 @@ def list_remote_jobs(tool_context: ToolContext, active_only: bool = False) -> di
     if active_only:
         jobs = [job for job in jobs if job["status"] not in TERMINAL_REMOTE_JOB_STATUSES]
     summaries = [
-        {key: job.get(key) for key in ("job_id", "provider", "node_id", "status", "external_id", "updated_at", "error")}
+        {
+            key: job.get(key)
+            for key in (
+                "job_id", "group_id", "provider", "node_id", "status",
+                "external_id", "updated_at", "error",
+            )
+        }
         for job in jobs
     ]
     return {"status": "ok", "job_count": len(summaries), "jobs": summaries}
@@ -486,7 +547,11 @@ def get_remote_job_status(job_id: str, tool_context: ToolContext) -> dict[str, A
     ):
         return {"status": "error", "message": "Remote job was not found in this session."}
     result = {
-        key: job[key] for key in ("job_id", "provider", "status", "external_id", "snapshot", "error", "updated_at")
+        key: job.get(key)
+        for key in (
+            "job_id", "group_id", "provider", "status", "external_id",
+            "snapshot", "error", "updated_at",
+        )
     }
     controls = [
         event["payload"] for event in service.store.list_events(job_id) if event["event_type"] == "user_control"

src/matcreator/agents/execution_agent/step_executor.py

@@ -19,6 +19,7 @@
 from .remote_job_tools import (
     attach_bohr_batchjob,
     collect_remote_job_outputs,
+    create_remote_job_group,
     download_remote_job_output,
     get_remote_job_status,
     pause_remote_job,
@@ -161,6 +162,10 @@ def _fill_missing_fields(self) -> "StepExecutorResult":
    job_id and error; never reinterpret its external ID or automatically submit a replacement.
 
 ## Choosing a remote-job submit tool
+- When submitting two or more similar or simultaneous Batch Jobs in this step,
+  ALWAYS call `create_remote_job_group` once first and pass its returned
+  `group_id` to every related `submit_bohr_batchjob` or `attach_bohr_batchjob`
+  call. Set `expected_jobs` to the exact number of jobs in that group.
 - `attach_bohr_batchjob`: track an already-submitted Batch Job using its explicit string
   `batchjob_id`. It only reads status and never submits. Use it for externally submitted
   jobs instead of creating a replacement; then use the returned `job_id` for status and outputs.
@@ -302,6 +307,7 @@ def build_step_executor_agent(llm_card: LLMCard) -> LlmAgent:
             FunctionTool(run_python),
             FunctionTool(run_bash),
             FunctionTool(submit_bohr_sandbox),
+            FunctionTool(create_remote_job_group),
             FunctionTool(submit_bohr_batchjob),
             FunctionTool(attach_bohr_batchjob),
             FunctionTool(get_remote_job_status),

src/matcreator/control_plane/providers/bohr_batchjob.py

@@ -21,8 +21,15 @@
     "failed": "failed",
     "deleted": "cancelled",
     "killed": "cancelled",
+    "stopped": "cancelled",
+    "terminated": "cancelled",
+    "cancelled": "cancelled",
+    "canceled": "cancelled",
 }
-_TERMINAL_STATUSES = {"succeeded", "failed", "deleted", "killed"}
+_TERMINAL_STATUSES = {
+    "succeeded", "failed", "deleted", "killed", "stopped", "terminated", "cancelled", "canceled",
+}
+_CANCELLED_STATUSES = {"deleted", "killed", "stopped", "terminated", "cancelled", "canceled"}
 _DOWNLOAD_TIMEOUT_SECONDS = 2 * 60 * 60 + 120
 
 
@@ -125,6 +132,8 @@ def status(self, external_id: str) -> RemoteJobStatus:
             "status_name": status_name or None,
             "terminal": status_name in _TERMINAL_STATUSES,
         }
+        if status_name in _CANCELLED_STATUSES:
+            snapshot["termination_origin"] = "provider_observed"
         for field in ("errorMessage", "errorCode"):
             value = data.get(field)
             if isinstance(value, (str, int, float)) and not isinstance(value, bool):

src/matcreator/control_plane/remote_job_monitor.py

@@ -56,6 +56,7 @@ async def run(self) -> None:
             while not self._stop.is_set():
                 try:
                     self._schedule_probes()
+                    self.store.evaluate_job_groups()
                     self._schedule_deliveries()
                 except Exception:
                     logger.exception("Remote job monitor scheduling failed; retrying next tick")
@@ -200,8 +201,10 @@ async def _deliver(self, notification_id: str) -> None:
     async def reconcile_once(self) -> list[dict[str, Any]]:
         """Wait for this bounded batch; the long-running loop never waits on a fleet."""
         tasks = self._schedule_probes()
+        self.store.evaluate_job_groups()
         self._schedule_deliveries()
         results = await asyncio.gather(*tasks)
+        self.store.evaluate_job_groups()
         self._schedule_deliveries()
         if self._deliveries:
             await asyncio.gather(*self._deliveries.values())

src/matcreator/control_plane/remote_job_service.py

@@ -84,6 +84,7 @@ def submit_job(
         step_number: int | None = None,
         output_dir: str | None = None,
         persisted_specification: dict[str, Any] | None = None,
+        group_id: str | None = None,
     ) -> dict[str, Any]:
         """Create one external job/sandbox once and persist its external ID.
 
@@ -115,6 +116,7 @@ def submit_job(
             step_number=step_number,
             specification=persisted_specification if persisted_specification is not None else spec,
             output_dir=output_dir,
+            group_id=group_id,
         )
         if job["status"] == "failed" and not job["external_id"]:
             # The previous attempt died before the provider handed back an
@@ -180,6 +182,7 @@ def attach_job(
         external_id: str,
         node_id: str | None = None,
         step_number: int | None = None,
+        group_id: str | None = None,
     ) -> dict[str, Any]:
         """Track an already-submitted job after a read-only status check; never submit.
 
@@ -200,6 +203,8 @@ def attach_job(
         ):
             raise ValueError("This remote job is already tracked in another session")
         if existing_jobs:
+            if existing_jobs[0].get("group_id") != group_id:
+                raise ValueError("This remote job is already tracked with different grouping")
             return self.reconcile_job(existing_jobs[0]["job_id"])
 
         adapter = self._adapter(provider)
@@ -224,6 +229,7 @@ def attach_job(
             node_id=node_id,
             step_number=step_number,
             specification={"attached": True},
+            group_id=group_id,
         )
         if job["external_id"]:
             return self.reconcile_job(job["job_id"])
@@ -245,6 +251,25 @@ def attach_job(
             expected_revision=job["state_revision"],
         )
 
+    def create_job_group(
+        self,
+        *,
+        owner_id: str,
+        session_id: str,
+        name: str,
+        expected_jobs: int,
+        failure_ratio: float | None = None,
+        deadline_seconds: float = 172800,
+    ) -> dict[str, Any]:
+        return self.store.create_job_group(
+            owner_id=owner_id,
+            session_id=session_id,
+            name=name,
+            expected_jobs=expected_jobs,
+            failure_ratio=failure_ratio,
+            deadline_seconds=deadline_seconds,
+        )
+
     def pause_job(self, job_id: str) -> dict[str, Any]:
         job = self._get_job(job_id)
         adapter = self._adapter(job["provider"])