diff --git a/docs/remote_job_monitoring.md b/docs/remote_job_monitoring.md
index 74ff9be..f92570d 100644
--- a/docs/remote_job_monitoring.md
+++ b/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
diff --git a/src/matcreator/agents/execution_agent/remote_job_tools.py b/src/matcreator/agents/execution_agent/remote_job_tools.py
index 069b62e..089f6b7 100644
--- a/src/matcreator/agents/execution_agent/remote_job_tools.py
+++ b/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"
diff --git a/src/matcreator/agents/execution_agent/step_executor.py b/src/matcreator/agents/execution_agent/step_executor.py
index c9062d2..9a2da7c 100644
--- a/src/matcreator/agents/execution_agent/step_executor.py
+++ b/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),
diff --git a/src/matcreator/control_plane/providers/bohr_batchjob.py b/src/matcreator/control_plane/providers/bohr_batchjob.py
index 902d234..b067d02 100644
--- a/src/matcreator/control_plane/providers/bohr_batchjob.py
+++ b/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):
diff --git a/src/matcreator/control_plane/remote_job_monitor.py b/src/matcreator/control_plane/remote_job_monitor.py
index cce037c..f2cd212 100644
--- a/src/matcreator/control_plane/remote_job_monitor.py
+++ b/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())
diff --git a/src/matcreator/control_plane/remote_job_service.py b/src/matcreator/control_plane/remote_job_service.py
index 4b9e06d..a904716 100644
--- a/src/matcreator/control_plane/remote_job_service.py
+++ b/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"])
diff --git a/src/matcreator/control_plane/remote_jobs.py b/src/matcreator/control_plane/remote_jobs.py
index 9d6c4a1..6f026e1 100644
--- a/src/matcreator/control_plane/remote_jobs.py
+++ b/src/matcreator/control_plane/remote_jobs.py
@@ -12,6 +12,10 @@
 TERMINAL_REMOTE_JOB_STATUSES = frozenset(
     {"collected", "failed", "cancelled", "terminated", "lost"}
 )
+GROUP_OUTCOME_STATUSES = frozenset(
+    {"succeeded", "collected", "failed", "cancelled", "terminated", "lost"}
+)
+GROUP_FAILURE_STATUSES = frozenset({"failed", "cancelled", "terminated", "lost"})
 ACTIVE_REMOTE_JOB_STATUSES = frozenset(
     {
         "created",
@@ -97,6 +101,7 @@ def _initialize(self) -> None:
                     artifacts TEXT NOT NULL DEFAULT '[]',
                     output_dir TEXT,
                     error TEXT,
+                    group_id TEXT,
                     state_revision INTEGER NOT NULL DEFAULT 0,
                     created_at REAL NOT NULL,
                     updated_at REAL NOT NULL
@@ -111,6 +116,21 @@ def _initialize(self) -> None:
                 CREATE INDEX IF NOT EXISTS idx_remote_jobs_external
                 ON remote_jobs(provider, external_id);
 
+                CREATE TABLE IF NOT EXISTS remote_job_groups (
+                    group_id TEXT PRIMARY KEY,
+                    owner_id TEXT NOT NULL,
+                    session_id TEXT NOT NULL,
+                    name TEXT NOT NULL,
+                    expected_jobs INTEGER NOT NULL,
+                    failure_ratio REAL,
+                    deadline_at REAL,
+                    delivery_reason TEXT,
+                    state_revision INTEGER NOT NULL DEFAULT 0,
+                    created_at REAL NOT NULL,
+                    updated_at REAL NOT NULL,
+                    UNIQUE(owner_id, session_id, name)
+                );
+
                 CREATE TABLE IF NOT EXISTS remote_job_events (
                     event_id INTEGER PRIMARY KEY AUTOINCREMENT,
                     job_id TEXT NOT NULL,
@@ -154,6 +174,14 @@ def _initialize(self) -> None:
                 ON remote_job_notifications(delivery_status, available_at, lease_until);
                 """
             )
+            columns = {
+                row["name"] for row in connection.execute("PRAGMA table_info(remote_jobs)").fetchall()
+            }
+            if "group_id" not in columns:
+                connection.execute("ALTER TABLE remote_jobs ADD COLUMN group_id TEXT")
+            connection.execute(
+                "CREATE INDEX IF NOT EXISTS idx_remote_jobs_group ON remote_jobs(group_id, status)"
+            )
 
     @staticmethod
     def _decode(row: sqlite3.Row | None) -> dict[str, Any] | None:
@@ -178,6 +206,7 @@ def create_job(
         step_number: int | None = None,
         specification: dict[str, Any] | None = None,
         output_dir: str | None = None,
+        group_id: str | None = None,
     ) -> dict[str, Any]:
         if not owner_id or not session_id or not provider or not idempotency_key:
             raise ValueError("owner_id, session_id, provider, and idempotency_key are required")
@@ -194,15 +223,31 @@ def create_job(
                     existing_data["owner_id"] != owner_id
                     or existing_data["session_id"] != session_id
                     or existing_data["provider"] != provider
+                    or existing_data.get("group_id") != group_id
                 ):
                     raise ValueError("Job idempotency key belongs to different work")
                 return existing_data
+            if group_id is not None:
+                group = connection.execute(
+                    "SELECT * FROM remote_job_groups WHERE group_id = ?", (group_id,)
+                ).fetchone()
+                if group is None:
+                    raise ValueError(f"Remote job group '{group_id}' was not found")
+                if group["owner_id"] != owner_id or group["session_id"] != session_id:
+                    raise ValueError("Remote job group belongs to a different session")
+                member_count = connection.execute(
+                    "SELECT COUNT(*) FROM remote_jobs WHERE group_id = ?", (group_id,)
+                ).fetchone()[0]
+                if member_count >= group["expected_jobs"]:
+                    raise ValueError("Remote job group already has its expected number of jobs")
+                if group["delivery_reason"] is not None:
+                    raise ValueError("Remote job group has already emitted its wakeup")
             connection.execute(
                 """
                 INSERT INTO remote_jobs (
                     job_id, owner_id, session_id, node_id, step_number, provider,
-                    idempotency_key, status, specification, output_dir, created_at, updated_at
-                ) VALUES (?, ?, ?, ?, ?, ?, ?, 'created', ?, ?, ?, ?)
+                    idempotency_key, status, specification, output_dir, group_id, created_at, updated_at
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, 'created', ?, ?, ?, ?, ?)
                 """,
                 (
                     job_id,
@@ -214,6 +259,7 @@ def create_job(
                     idempotency_key,
                     json.dumps(specification or {}, sort_keys=True),
                     output_dir,
+                    group_id,
                     now,
                     now,
                 ),
@@ -221,6 +267,85 @@ def create_job(
             self._append_event(connection, job_id, "created", {"status": "created"}, now)
         return self.get_job(job_id) or {}
 
+    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]:
+        if not owner_id or not session_id or not isinstance(name, str) or not name.strip():
+            raise ValueError("owner_id, session_id, and a nonempty group name are required")
+        if isinstance(expected_jobs, bool) or not isinstance(expected_jobs, int) or expected_jobs < 1:
+            raise ValueError("expected_jobs must be a positive integer")
+        if failure_ratio is not None and (
+            isinstance(failure_ratio, bool)
+            or not isinstance(failure_ratio, (int, float))
+            or not 0 < float(failure_ratio) <= 1
+        ):
+            raise ValueError("failure_ratio must be greater than 0 and at most 1")
+        if (
+            isinstance(deadline_seconds, bool)
+            or not isinstance(deadline_seconds, (int, float))
+            or float(deadline_seconds) <= 0
+        ):
+            raise ValueError("deadline_seconds must be positive")
+        now = time.time()
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            existing = connection.execute(
+                """SELECT * FROM remote_job_groups
+                   WHERE owner_id = ? AND session_id = ? AND name = ?""",
+                (owner_id, session_id, name.strip()),
+            ).fetchone()
+            deadline_at = now + float(deadline_seconds)
+            if existing is not None:
+                existing_deadline_seconds = existing["deadline_at"] - existing["created_at"]
+                if (
+                    existing["expected_jobs"] != expected_jobs
+                    or existing["failure_ratio"] != (
+                        float(failure_ratio) if failure_ratio is not None else None
+                    )
+                    or abs(existing_deadline_seconds - float(deadline_seconds)) > 0.001
+                ):
+                    raise ValueError("Remote job group name already has a different policy")
+                return dict(existing)
+            group_id = uuid.uuid4().hex
+            connection.execute(
+                """INSERT INTO remote_job_groups (
+                   group_id, owner_id, session_id, name, expected_jobs, failure_ratio,
+                   deadline_at, created_at, updated_at
+                   ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                (
+                    group_id, owner_id, session_id, name.strip(), expected_jobs,
+                    float(failure_ratio) if failure_ratio is not None else None,
+                    deadline_at, now, now,
+                ),
+            )
+            row = connection.execute(
+                "SELECT * FROM remote_job_groups WHERE group_id = ?", (group_id,)
+            ).fetchone()
+        return dict(row)
+
+    def get_job_group(self, group_id: str) -> dict[str, Any] | None:
+        with self._connect() as connection:
+            row = connection.execute(
+                "SELECT * FROM remote_job_groups WHERE group_id = ?", (group_id,)
+            ).fetchone()
+        return dict(row) if row is not None else None
+
+    def list_job_groups(self, *, owner_id: str, session_id: str) -> list[dict[str, Any]]:
+        with self._connect() as connection:
+            rows = connection.execute(
+                """SELECT * FROM remote_job_groups WHERE owner_id = ? AND session_id = ?
+                   ORDER BY created_at DESC""",
+                (owner_id, session_id),
+            ).fetchall()
+        return [dict(row) for row in rows]
+
     def get_job(self, job_id: str) -> dict[str, Any] | None:
         with self._connect() as connection:
             row = connection.execute("SELECT * FROM remote_jobs WHERE job_id = ?", (job_id,)).fetchone()
@@ -311,16 +436,101 @@ def transition_job(
                 {"from": current["status"], "to": status},
                 now,
             )
-            if status != current["status"] and status in {"succeeded", "failed", "cancelled", "lost"}:
-                self._enqueue_notification(
-                    connection,
-                    {**current, "external_id": resulting_external_id, "status": status,
-                     "state_revision": current["state_revision"] + 1, "snapshot": merged_snapshot,
-                     "error": current["error"] if error is _UNSET else error},
-                    kind="lifecycle", now=now,
-                )
+            if status != current["status"]:
+                transitioned = {
+                    **current,
+                    "external_id": resulting_external_id,
+                    "status": status,
+                    "state_revision": current["state_revision"] + 1,
+                    "snapshot": merged_snapshot,
+                    "error": current["error"] if error is _UNSET else error,
+                }
+                if status in {"succeeded", "failed", "cancelled", "lost"} and not current.get("group_id"):
+                    self._enqueue_notification(connection, transitioned, kind="lifecycle", now=now)
+                if current.get("group_id"):
+                    self._evaluate_job_group(connection, current["group_id"], now=now)
         return self.get_job(job_id) or {}
 
+    def evaluate_job_groups(self) -> int:
+        """Evaluate pending deadlines and return the number of newly queued group wakeups."""
+        now = time.time()
+        queued = 0
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            rows = connection.execute(
+                "SELECT group_id FROM remote_job_groups WHERE delivery_reason IS NULL"
+            ).fetchall()
+            for row in rows:
+                queued += int(self._evaluate_job_group(connection, row["group_id"], now=now))
+        return queued
+
+    @staticmethod
+    def _evaluate_job_group(
+        connection: sqlite3.Connection, group_id: str, *, now: float,
+    ) -> bool:
+        group = connection.execute(
+            "SELECT * FROM remote_job_groups WHERE group_id = ?", (group_id,)
+        ).fetchone()
+        if group is None or group["delivery_reason"] is not None:
+            return False
+        jobs = connection.execute(
+            """SELECT job_id, provider, external_id, node_id, status, error, snapshot
+               FROM remote_jobs WHERE group_id = ? ORDER BY created_at, job_id""",
+            (group_id,),
+        ).fetchall()
+        if not jobs:
+            return False
+        expected = group["expected_jobs"]
+        outcomes = sum(job["status"] in GROUP_OUTCOME_STATUSES for job in jobs)
+        failures = sum(job["status"] in GROUP_FAILURE_STATUSES for job in jobs)
+        reason = None
+        if (
+            len(jobs) == expected
+            and group["failure_ratio"] is not None
+            and failures / expected >= group["failure_ratio"]
+        ):
+            reason = "failure_ratio"
+        elif len(jobs) == expected and outcomes == expected:
+            reason = "all_terminal"
+        elif group["deadline_at"] is not None and now >= group["deadline_at"]:
+            reason = "deadline"
+        if reason is None:
+            return False
+        revision = group["state_revision"] + 1
+        updated = connection.execute(
+            """UPDATE remote_job_groups SET delivery_reason = ?, state_revision = ?, updated_at = ?
+               WHERE group_id = ? AND delivery_reason IS NULL AND state_revision = ?""",
+            (reason, revision, now, group_id, group["state_revision"]),
+        )
+        if updated.rowcount != 1:
+            return False
+        summary = []
+        for job in jobs:
+            item = dict(job)
+            item["snapshot"] = json.loads(item["snapshot"])
+            summary.append(item)
+        anchor = jobs[-1]
+        payload = {
+            "group_id": group_id,
+            "group_name": group["name"],
+            "group_reason": reason,
+            "expected_jobs": expected,
+            "outcome_jobs": outcomes,
+            "failed_jobs": failures,
+            "jobs": summary,
+        }
+        connection.execute(
+            """INSERT OR IGNORE INTO remote_job_notifications (
+               notification_id, job_id, owner_id, session_id, state_revision, kind, status,
+               payload, available_at, created_at, updated_at
+               ) VALUES (?, ?, ?, ?, ?, 'group', ?, ?, ?, ?, ?)""",
+            (
+                uuid.uuid4().hex, anchor["job_id"], group["owner_id"], group["session_id"],
+                revision, reason, json.dumps(payload, sort_keys=True), now, now, now,
+            ),
+        )
+        return True
+
     def reset_failed_job_for_retry(self, job_id: str) -> dict[str, Any]:
         """Return a failed job that never acquired an external ID to ``created``.
 
@@ -466,26 +676,36 @@ def record_user_control(self, job_id: str, action: str) -> None:
             connection.execute(
                 """UPDATE remote_job_notifications SET delivery_status = 'suppressed',
                    last_error = 'explicit user control', claim_token = NULL, lease_until = NULL,
-                   updated_at = ? WHERE job_id = ? AND delivery_status IN ('pending', 'claimed')""",
+                   updated_at = ? WHERE job_id = ? AND kind != 'group'
+                   AND delivery_status IN ('pending', 'claimed')""",
                 (time.time(), job_id),
             )
 
     def notifications_suppressed(
-        self, owner_id: str, session_id: str, *, job_id: str | None = None,
+        self,
+        owner_id: str,
+        session_id: str,
+        *,
+        job_id: str | None = None,
+        include_user_control: bool = True,
     ) -> bool:
         """Check a stop marker, or a specific job's cutoff and explicit controls."""
         with self._connect() as connection:
             if job_id is not None:
-                return connection.execute(
+                stop = connection.execute(
                     """SELECT 1 FROM remote_job_notification_stops s JOIN remote_jobs j
                        ON j.owner_id = s.owner_id AND j.session_id = s.session_id
                        WHERE s.owner_id = ? AND s.session_id = ? AND j.job_id = ?
-                         AND j.created_at <= s.stopped_at
-                       UNION ALL
-                       SELECT 1 FROM remote_job_events e JOIN remote_jobs j ON j.job_id = e.job_id
+                         AND j.created_at <= s.stopped_at LIMIT 1""",
+                    (owner_id, session_id, job_id),
+                ).fetchone()
+                if stop is not None or not include_user_control:
+                    return stop is not None
+                return connection.execute(
+                    """SELECT 1 FROM remote_job_events e JOIN remote_jobs j ON j.job_id = e.job_id
                        WHERE j.owner_id = ? AND j.session_id = ? AND j.job_id = ?
                          AND e.event_type = 'user_control' LIMIT 1""",
-                    (owner_id, session_id, job_id, owner_id, session_id, job_id),
+                    (owner_id, session_id, job_id),
                 ).fetchone() is not None
             return connection.execute(
                 "SELECT 1 FROM remote_job_notification_stops WHERE owner_id = ? AND session_id = ?",
@@ -565,7 +785,9 @@ def claim_notification(self, notification_id: str, *, lease_seconds: float = 60)
                          AND j.created_at <= s.stopped_at)
                    AND NOT EXISTS (
                        SELECT 1 FROM remote_job_events e
-                       WHERE e.job_id = remote_job_notifications.job_id AND e.event_type = 'user_control')""",
+                       WHERE remote_job_notifications.kind != 'group'
+                         AND e.job_id = remote_job_notifications.job_id
+                         AND e.event_type = 'user_control')""",
                 (token, now + lease_seconds, now, notification_id, now, now),
             )
             if updated.rowcount != 1:
diff --git a/src/matcreator/skills/remote-job/SKILL.md b/src/matcreator/skills/remote-job/SKILL.md
index 175cdb8..0d5db1c 100644
--- a/src/matcreator/skills/remote-job/SKILL.md
+++ b/src/matcreator/skills/remote-job/SKILL.md
@@ -4,6 +4,7 @@ description: Submit, monitor, and control tracked remote jobs on the Bohrium pla
 metadata:
   tools:
     - submit_bohr_sandbox
+    - create_remote_job_group
     - submit_bohr_batchjob
     - attach_bohr_batchjob
     - get_remote_job_status
@@ -53,6 +54,12 @@ new compute. A raw provider ID alone is not a tracked job.
   Optional: `out_files` (list of retained paths), `max_run_time="24h"`,
   `max_wait_time="30m"` (duration strings, not numeric seconds).
 
+  For several independent Batch Jobs whose outcomes should be handled together,
+  call `create_remote_job_group` first with the exact `expected_jobs`, then pass
+  its `group_id` to every submission. Grouped jobs suppress individual lifecycle
+  wakeups and emit one aggregate wakeup when all expected jobs have outcomes, or
+  when the deadline (48 hours by default) or optional failed-job ratio is reached.
+
 ## Sandbox lifecycle
 
 1. Call `submit_bohr_sandbox` once for the current step and record the
diff --git a/tests/test_bohr_batchjob_adapter.py b/tests/test_bohr_batchjob_adapter.py
index c911fe0..52f8beb 100644
--- a/tests/test_bohr_batchjob_adapter.py
+++ b/tests/test_bohr_batchjob_adapter.py
@@ -287,6 +287,8 @@ def fake_run(command, **kwargs):
     ("active", "running", False), ("running", "running", False),
     ("succeeded", "succeeded", True), ("failed", "failed", True),
     ("deleted", "cancelled", True), ("killed", "cancelled", True),
+    ("stopped", "cancelled", True), ("terminated", "cancelled", True),
+    ("cancelled", "cancelled", True), ("canceled", "cancelled", True),
     ("unknown", None, False), ("future-status", None, False),
     ("", None, False),
 ])
@@ -301,11 +303,14 @@ def fake_run(command, **kwargs):
     monkeypatch.setattr(subprocess, "run", fake_run)
     status = BohrBatchJobAdapter().status("batch-job-123")
     assert status.normalized_status == normalized
-    assert status.snapshot == {
+    expected = {
         "provider_status": name or "unknown",
         "status_name": name or None,
         "terminal": terminal,
     }
+    if normalized == "cancelled":
+        expected["termination_origin"] = "provider_observed"
+    assert status.snapshot == expected
 
 
 @pytest.mark.parametrize("name,normalized", [
@@ -321,6 +326,7 @@ def test_terminal_snapshot_matches_semantic_outcome(monkeypatch, name, normalize
     assert status.normalized_status == normalized
     assert status.snapshot == {
         "provider_status": name, "status_name": name, "terminal": True,
+        **({"termination_origin": "provider_observed"} if normalized == "cancelled" else {}),
     }
 
 
@@ -391,6 +397,7 @@ def test_status_observed_bohr_2_6_86_describe_shape(
         "terminal": terminal,
         "errorCode": 0,
         "errorMessage": data["errorMessage"],
+        **({"termination_origin": "provider_observed"} if normalized == "cancelled" else {}),
     }
 
 
diff --git a/tests/test_remote_job_tools.py b/tests/test_remote_job_tools.py
index ac62ee2..4a1610b 100644
--- a/tests/test_remote_job_tools.py
+++ b/tests/test_remote_job_tools.py
@@ -6,6 +6,7 @@
 from google.adk.tools.function_tool import FunctionTool
 
 from matcreator.agents.execution_agent import remote_job_tools
+from matcreator.agents.execution_agent.step_executor import _STEP_EXECUTOR_INSTRUCTION
 from matcreator.control_plane.providers.e2b import E2BConnectionConfig
 
 
@@ -19,6 +20,7 @@ def submit_job(self, **kwargs):
         self.submissions.append(kwargs)
         return {
             "job_id": "job-123",
+            "group_id": None,
             "status": "running",
             "external_id": "sandbox-123",
         }
@@ -265,7 +267,7 @@ def test_submit_bohr_batchjob_exposes_the_expected_adk_schema() -> None:
     assert set(parameters["required"]) == {"name", "image", "command"}
     assert set(parameters["properties"]) == {
         "name", "image", "command", "project_id", "machine_type", "sku_id",
-        "input_path", "out_files", "max_run_time", "max_wait_time",
+        "input_path", "out_files", "max_run_time", "max_wait_time", "group_id",
     }
 
 
@@ -276,7 +278,7 @@ def test_attach_bohr_batchjob_exposes_required_string_id() -> None:
     parameters = payload.get("parameters_json_schema") or payload.get("parameters")
     assert payload["name"] == "attach_bohr_batchjob"
     assert parameters["required"] == ["batchjob_id"]
-    assert set(parameters["properties"]) == {"batchjob_id"}
+    assert set(parameters["properties"]) == {"batchjob_id", "group_id"}
     assert parameters["properties"]["batchjob_id"]["type"].lower() == "string"
 
 
@@ -299,6 +301,7 @@ def attach(**kwargs):
     assert calls == [{
         "owner_id": "alice", "session_id": "session-1", "provider": "bohr_batchjob",
         "external_id": "external-123", "node_id": "relax", "step_number": 2,
+        "group_id": None,
     }]
     assert references == [{
         "session_id": "session-1", "node_id": "relax", "job_id": "job-123",
@@ -363,6 +366,49 @@ def test_submit_bohr_batchjob_tool_submits_batch_spec(monkeypatch) -> None:
     assert submission["spec"]["max_wait_time"] == "30m"
 
 
+def test_create_remote_job_group_exposes_policy_and_context(monkeypatch) -> None:
+    calls = []
+
+    def create(**kwargs):
+        calls.append(kwargs)
+        return {
+            "group_id": "group-123", "name": "sweep", "expected_jobs": 3,
+            "failure_ratio": 0.5, "deadline_at": 1000,
+        }
+
+    monkeypatch.setattr(
+        remote_job_tools, "_service", lambda: SimpleNamespace(create_job_group=create),
+    )
+    result = remote_job_tools.create_remote_job_group(
+        _context(), name="sweep", expected_jobs=3, failure_ratio=0.5, deadline_seconds=600,
+    )
+
+    assert result["status"] == "ready"
+    assert result["group_id"] == "group-123"
+    assert calls == [{
+        "owner_id": "alice", "session_id": "session-1", "name": "sweep",
+        "expected_jobs": 3, "failure_ratio": 0.5, "deadline_seconds": 600,
+    }]
+
+
+def test_parallel_batch_job_tools_require_one_shared_group() -> None:
+    assert "ALWAYS call `create_remote_job_group` once first" in _STEP_EXECUTOR_INSTRUCTION
+    assert "exact number of jobs in that group" in _STEP_EXECUTOR_INSTRUCTION
+
+    group_description = FunctionTool(
+        remote_job_tools.create_remote_job_group
+    )._get_declaration().description
+    submit_description = FunctionTool(
+        remote_job_tools.submit_bohr_batchjob
+    )._get_declaration().description
+    attach_description = FunctionTool(
+        remote_job_tools.attach_bohr_batchjob
+    )._get_declaration().description
+    assert "ALWAYS call this once" in group_description
+    assert "similar or simultaneous Batch Jobs" in submit_description
+    assert "two or more related Batch Jobs" in attach_description
+
+
 @pytest.mark.parametrize("selectors", [{}, {"machine_type": "cpu", "sku_id": 123}])
 def test_submit_bohr_batchjob_requires_exactly_one_machine(monkeypatch, selectors) -> None:
     service = _FakeService()
@@ -742,6 +788,7 @@ def test_list_remote_jobs_returns_compact_projection_for_current_session(monkeyp
         },
         {
             "job_id": "job-456",
+            "group_id": None,
             "owner_id": "alice",
             "session_id": "session-1",
             "provider": "bohr_batchjob",
@@ -772,6 +819,7 @@ def test_list_remote_jobs_returns_compact_projection_for_current_session(monkeyp
     assert result["jobs"] == [
         {
             "job_id": "job-123",
+            "group_id": None,
             "provider": "e2b",
             "node_id": "relax",
             "status": "running",
@@ -781,6 +829,7 @@ def test_list_remote_jobs_returns_compact_projection_for_current_session(monkeyp
         },
         {
             "job_id": "job-456",
+            "group_id": None,
             "provider": "bohr_batchjob",
             "node_id": "scf",
             "status": "collected",
diff --git a/tests/test_remote_jobs.py b/tests/test_remote_jobs.py
index 79aecc2..66a950c 100644
--- a/tests/test_remote_jobs.py
+++ b/tests/test_remote_jobs.py
@@ -235,6 +235,99 @@ def fail(*args, **kwargs):
     assert store.list_events(job["job_id"])[-1]["payload"]["to"] == "running"
 
 
+def test_job_group_suppresses_member_wakeups_until_all_outcomes(tmp_path):
+    store = RemoteJobStore(tmp_path / "jobs.db")
+    group = store.create_job_group(
+        owner_id="alice", session_id="s", name="parameter-sweep", expected_jobs=2,
+    )
+    first = store.create_job(
+        owner_id="alice", session_id="s", provider="bohr_batchjob",
+        idempotency_key="first", group_id=group["group_id"],
+    )
+    second = store.create_job(
+        owner_id="alice", session_id="s", provider="bohr_batchjob",
+        idempotency_key="second", group_id=group["group_id"],
+    )
+    for job, external_id in ((first, "batch-1"), (second, "batch-2")):
+        store.transition_job(job["job_id"], "submitting")
+        store.transition_job(job["job_id"], "running", external_id=external_id)
+
+    store.transition_job(first["job_id"], "succeeded")
+    assert store.list_pending_notifications() == []
+    store.transition_job(second["job_id"], "failed", error="queue timeout")
+
+    notification, = store.list_pending_notifications()
+    assert notification["kind"] == "group"
+    assert notification["status"] == notification["group_reason"] == "all_terminal"
+    assert notification["outcome_jobs"] == 2
+    assert notification["failed_jobs"] == 1
+    assert [job["status"] for job in notification["jobs"]] == ["succeeded", "failed"]
+    assert store.get_job_group(group["group_id"])["delivery_reason"] == "all_terminal"
+
+
+def test_job_group_wakes_once_on_failure_ratio(tmp_path):
+    store = RemoteJobStore(tmp_path / "jobs.db")
+    group = store.create_job_group(
+        owner_id="alice", session_id="s", name="screening", expected_jobs=1,
+        failure_ratio=0.25,
+    )
+    job = store.create_job(
+        owner_id="alice", session_id="s", provider="bohr_batchjob",
+        idempotency_key="failed-member", group_id=group["group_id"],
+    )
+    store.transition_job(job["job_id"], "submitting")
+    store.transition_job(job["job_id"], "running", external_id="batch-1")
+    store.transition_job(job["job_id"], "cancelled")
+    store.evaluate_job_groups()
+
+    notification, = store.list_pending_notifications()
+    assert notification["group_reason"] == "failure_ratio"
+    assert len(store.list_notifications()) == 1
+
+
+def test_job_group_deadline_is_evaluated_by_monitor_tick(tmp_path, monkeypatch):
+    clock = [100.0]
+    monkeypatch.setattr("matcreator.control_plane.remote_jobs.time.time", lambda: clock[0])
+    store = RemoteJobStore(tmp_path / "jobs.db")
+    group = store.create_job_group(
+        owner_id="alice", session_id="s", name="deadline", expected_jobs=2,
+        deadline_seconds=10,
+    )
+    job = store.create_job(
+        owner_id="alice", session_id="s", provider="bohr_batchjob",
+        idempotency_key="running-member", group_id=group["group_id"],
+    )
+    store.transition_job(job["job_id"], "submitting")
+    store.transition_job(job["job_id"], "running", external_id="batch-1")
+    clock[0] = 111.0
+
+    assert store.evaluate_job_groups() == 1
+    notification, = store.list_pending_notifications()
+    assert notification["group_reason"] == "deadline"
+    assert notification["outcome_jobs"] == 0
+
+
+def test_job_group_validates_membership_and_policy(tmp_path):
+    store = RemoteJobStore(tmp_path / "jobs.db")
+    group = store.create_job_group(
+        owner_id="alice", session_id="s", name="one", expected_jobs=1,
+    )
+    store.create_job(
+        owner_id="alice", session_id="s", provider="bohr_batchjob",
+        idempotency_key="member", group_id=group["group_id"],
+    )
+    with pytest.raises(ValueError, match="expected number"):
+        store.create_job(
+            owner_id="alice", session_id="s", provider="bohr_batchjob",
+            idempotency_key="extra", group_id=group["group_id"],
+        )
+    with pytest.raises(ValueError, match="different session"):
+        store.create_job(
+            owner_id="bob", session_id="other", provider="bohr_batchjob",
+            idempotency_key="foreign", group_id=group["group_id"],
+        )
+
+
 def test_session_stop_cutoff_allows_newly_approved_jobs_after_restart(tmp_path, monkeypatch):
     clock = [100.0]
     monkeypatch.setattr("matcreator.control_plane.remote_jobs.time.time", lambda: clock[0])
diff --git a/tests/test_web_remote_job_wakeups.py b/tests/test_web_remote_job_wakeups.py
index 7ec9f9c..8c88737 100644
--- a/tests/test_web_remote_job_wakeups.py
+++ b/tests/test_web_remote_job_wakeups.py
@@ -78,6 +78,42 @@ async def produce(run, payload, target_url, *, started):
     asyncio.run(exercise())
 
 
+def test_group_notification_contains_aggregate_outcomes(harness, monkeypatch):
+    harness.store.suppress_notification(harness.notification["notification_id"])
+    group = harness.store.create_job_group(
+        owner_id="alice", session_id="session-1", name="sweep", expected_jobs=1,
+    )
+    job = harness.store.create_job(
+        owner_id="alice", session_id="session-1", node_id="screen",
+        provider="bohr_batchjob", idempotency_key="grouped-job",
+        specification={}, group_id=group["group_id"],
+    )
+    harness.store.transition_job(job["job_id"], "submitting")
+    harness.store.transition_job(job["job_id"], "succeeded", external_id="batch-grouped")
+    notification, = harness.store.list_pending_notifications()
+
+    async def exercise():
+        captured = []
+        monkeypatch.setattr(harness.web, "_target_url_for_user", _target)
+
+        async def produce(run, payload, target_url, *, started):
+            captured.append(payload)
+            await harness.web._run_registry.publish(run, 'data: {"author":"agent"}\n\n')
+            started.set()
+
+        monkeypatch.setattr(harness.web, "_produce_managed_run", produce)
+        run_id = await harness.web._resume_remote_job_session(notification)
+        assert run_id
+        text = captured[0]["new_message"]["parts"][0]["text"]
+        assert f"Remote job group: sweep ({group['group_id']})" in text
+        assert "Wakeup reason: all_terminal" in text
+        assert f"- {job['job_id']}: succeeded (bohr_batchjob)" in text
+        assert "Call list_remote_jobs" in text
+        await harness.web._run_registry.shutdown()
+
+    asyncio.run(exercise())
+
+
 async def _target(owner_id):
     assert owner_id == "alice"
     return "http://worker.example"
diff --git a/web/main.py b/web/main.py
index 901402e..f7156e3 100644
--- a/web/main.py
+++ b/web/main.py
@@ -1264,11 +1264,16 @@ async def _resume_remote_job_session(notification: dict[str, Any]) -> str | None
     """Start an ordinary managed turn for one durable completion notification."""
     owner_id = notification["owner_id"]
     session_id = notification["session_id"]
+    is_group = notification.get("kind") == "group"
     store = _remote_job_store_for_owner(owner_id)
     if _remote_job_monitor_stop.is_set():
         return None
     if await asyncio.to_thread(
-        store.notifications_suppressed, owner_id, session_id, job_id=notification["job_id"],
+        store.notifications_suppressed,
+        owner_id,
+        session_id,
+        job_id=notification["job_id"],
+        include_user_control=not is_group,
     ):
         await asyncio.to_thread(store.suppress_notification, notification["notification_id"], "Session stopped")
         return None
@@ -1298,25 +1303,45 @@ async def _resume_remote_job_session(notification: dict[str, Any]) -> str | None
     if job["owner_id"] != owner_id or job["session_id"] != session_id:
         raise ValueError("Remote job notification does not match its owning session")
     controls = await asyncio.to_thread(store.list_events, job["job_id"])
-    if any(event["event_type"] == "user_control" for event in controls):
+    if not is_group and any(event["event_type"] == "user_control" for event in controls):
         await asyncio.to_thread(store.suppress_notification, notification["notification_id"], "Job stopped by user")
         return None
-    message = (
-        "REMOTE JOB STATUS UPDATE from the harness.\n"
-        f"Notification ID: {notification['notification_id']}\n"
-        f"Tracked job_id: {job['job_id']}\n"
-        f"Graph node: {job.get('node_id')}\n"
-        f"Provider: {job['provider']}\n"
-        f"Event kind: {notification.get('kind', 'lifecycle')}\n"
-        f"Observed outcome: {notification['status']}; current allocation status: {job['status']}\n"
-        "Read get_remote_job_status with this job_id first. Process the results of "
-        "this already-submitted work: collect and validate available outputs on success, "
-        "or inspect and report failure/timeout. If a sandbox background command finished, "
-        "inspect its persisted result without rerunning it. Reuse already-collected artifacts. "
-        "Respect user controls and any newer session instructions. Do not submit a replacement "
-        "job, repeat the computation, or authorize additional compute automatically. "
-        "If this notification has already been handled, do not repeat its side effects."
-    )
+    if is_group:
+        group_jobs = notification.get("jobs") or []
+        job_lines = "\n".join(
+            f"- {item.get('job_id')}: {item.get('status')} ({item.get('provider')})"
+            for item in group_jobs
+        )
+        message = (
+            "REMOTE JOB STATUS UPDATE from the harness.\n"
+            f"Notification ID: {notification['notification_id']}\n"
+            f"Remote job group: {notification.get('group_name')} ({notification.get('group_id')})\n"
+            f"Wakeup reason: {notification.get('group_reason')}\n"
+            f"Outcomes: {notification.get('outcome_jobs')}/{notification.get('expected_jobs')}; "
+            f"failed/cancelled/lost: {notification.get('failed_jobs')}\n"
+            f"Jobs:\n{job_lines}\n"
+            "Call list_remote_jobs, then read get_remote_job_status for every listed job before "
+            "processing this already-submitted work. Collect and validate successful outputs and "
+            "inspect failures or timeouts. Respect user controls and newer session instructions. "
+            "Do not resubmit jobs or authorize additional compute automatically."
+        )
+    else:
+        message = (
+            "REMOTE JOB STATUS UPDATE from the harness.\n"
+            f"Notification ID: {notification['notification_id']}\n"
+            f"Tracked job_id: {job['job_id']}\n"
+            f"Graph node: {job.get('node_id')}\n"
+            f"Provider: {job['provider']}\n"
+            f"Event kind: {notification.get('kind', 'lifecycle')}\n"
+            f"Observed outcome: {notification['status']}; current allocation status: {job['status']}\n"
+            "Read get_remote_job_status with this job_id first. Process the results of "
+            "this already-submitted work: collect and validate available outputs on success, "
+            "or inspect and report failure/timeout. If a sandbox background command finished, "
+            "inspect its persisted result without rerunning it. Reuse already-collected artifacts. "
+            "Respect user controls and any newer session instructions. Do not submit a replacement "
+            "job, repeat the computation, or authorize additional compute automatically. "
+            "If this notification has already been handled, do not repeat its side effects."
+        )
     payload = {
         "app_name": APP_NAME,
         "user_id": owner_id,
@@ -1330,7 +1355,11 @@ async def still_allowed() -> bool:
         if _remote_job_monitor_stop.is_set():
             return False
         if await asyncio.to_thread(
-            store.notifications_suppressed, owner_id, session_id, job_id=job["job_id"],
+            store.notifications_suppressed,
+            owner_id,
+            session_id,
+            job_id=job["job_id"],
+            include_user_control=not is_group,
         ) or is_cancellation_requested(session_id, workspace_root=cancellation_root):
             await asyncio.to_thread(store.suppress_notification, notification["notification_id"], "Session stopped")
             return False
@@ -3276,6 +3305,7 @@ async def list_session_remote_jobs(
             "jobs": store.list_jobs(
                 owner_id=user_id, session_id=session_id
             ),
+            "groups": store.list_job_groups(owner_id=user_id, session_id=session_id),
             "active_run": active_run.summary() if active_run else None,
             "activity_revision": hashlib.sha256(activity_revision.encode()).hexdigest()
             if notifications or latest_run else None,
diff --git a/web/vite-frontend/src/features/remoteJobs/RemoteJobsController.js b/web/vite-frontend/src/features/remoteJobs/RemoteJobsController.js
index b9c5642..0319ae0 100644
--- a/web/vite-frontend/src/features/remoteJobs/RemoteJobsController.js
+++ b/web/vite-frontend/src/features/remoteJobs/RemoteJobsController.js
@@ -1151,6 +1151,7 @@ export function createRemoteJobsController({
   function renderPresentationJobs() {
     if (presentationJobs === null) return false;
     state.remoteJobs = clonePresentationJobs();
+    state.remoteJobGroups = [];
     render();
     onJobsChanged();
     return true;
@@ -1226,12 +1227,16 @@ export function createRemoteJobsController({
     }
 
     const currentJobIds = new Set();
+    const groupsById = new Map(
+      (state.remoteJobGroups || []).map((group) => [String(group.group_id), group]),
+    );
     state.remoteJobs.forEach((job, index) => {
       const item = documentRef.createElement("li");
       const lifecycle = remoteJobLifecycle(projectedLifecycleKey(job));
       const presentation = normalizeRemoteJobPresentation(job);
       const jobKey = String(job.job_id || job.external_id || `remote-job-${index}`);
       const jobLabel = String(job.external_id || job.job_id || "remote job");
+      const group = job.group_id ? groupsById.get(String(job.group_id)) : null;
       const summaryId = `remote-job-summary-${index}`;
       const detailsId = `remote-job-details-${index}`;
       currentJobIds.add(jobKey);
@@ -1278,6 +1283,13 @@ export function createRemoteJobsController({
       identifier.textContent = job.external_id || job.job_id || "—";
       identifier.title = String(job.external_id || job.job_id || "—");
       identity.append(identityLabel, identifier);
+      if (job.group_id) {
+        const groupBadge = documentRef.createElement("span");
+        groupBadge.className = "remote-job-group";
+        groupBadge.textContent = group?.name ? `Group · ${group.name}` : "Grouped job";
+        groupBadge.title = `Group ID: ${job.group_id}`;
+        identity.appendChild(groupBadge);
+      }
       const statusRow = documentRef.createElement("div");
       statusRow.className = "remote-job-status-row";
       const status = documentRef.createElement("span");
@@ -1339,6 +1351,16 @@ export function createRemoteJobsController({
       const detailRows = documentRef.createElement("div");
       detailRows.className = "remote-job-detail-rows";
       detailRows.appendChild(createDetailRow("Job ID", job.job_id));
+      if (job.group_id) {
+        detailRows.appendChild(createDetailRow("Group ID", job.group_id));
+        if (group?.name) detailRows.appendChild(createDetailRow("Group name", group.name));
+        if (group?.expected_jobs) {
+          detailRows.appendChild(createDetailRow("Expected jobs", group.expected_jobs));
+        }
+        if (group?.delivery_reason) {
+          detailRows.appendChild(createDetailRow("Group wakeup", group.delivery_reason));
+        }
+      }
       detailRows.appendChild(createDetailRow(
         SANDBOX_PROVIDERS.has(String(job.provider || "").toLowerCase()) ? "Sandbox ID" : "Provider ID",
         job.external_id,
@@ -1400,6 +1422,7 @@ export function createRemoteJobsController({
       if (renderPresentationJobs()) return;
       if (sessionId !== state.sessionId || owner !== state.activeSessionUserId) return;
       state.remoteJobs = Array.isArray(data?.jobs) ? data.jobs : [];
+      state.remoteJobGroups = Array.isArray(data?.groups) ? data.groups : [];
       lastActivityHadActiveRun = Boolean(data?.active_run);
       render();
       onJobsChanged({ sessionId, owner, activity: data });
@@ -1476,6 +1499,7 @@ export function createRemoteJobsController({
   function reset({ notify = false } = {}) {
     stopPolling();
     state.remoteJobs = presentationJobs === null ? [] : clonePresentationJobs();
+    state.remoteJobGroups = [];
     lastActivityHadActiveRun = false;
     wasActive = false;
     graceRemainingMs = 0;
diff --git a/web/vite-frontend/src/features/session/runtime.js b/web/vite-frontend/src/features/session/runtime.js
index acb7018..d305a25 100644
--- a/web/vite-frontend/src/features/session/runtime.js
+++ b/web/vite-frontend/src/features/session/runtime.js
@@ -51,6 +51,12 @@ export function managedRecoverySource(activeRun = {}) {
     : "replay";
 }
 
+function managedEventKey(event) {
+  const explicit = event?.id ?? event?.event_id ?? event?.eventId;
+  if (explicit !== undefined && explicit !== null && String(explicit)) return `id:${explicit}`;
+  return `event:${JSON.stringify(event)}`;
+}
+
 export function shouldShowApproval(sessionId, sessionData, events, options = {}) {
   const normalized = typeof options === "function" ? { isSuppressed: options } : (options || {});
   const suppressedPlanApprovalTurns = normalized.suppressedPlanApprovalTurns || new Map();
@@ -713,6 +719,7 @@ export function createSessionRuntime({
       message, messageView: view,
       managedPresentation: {
         message, shownPlots, view, scheduler, lineBuffer: "", hydratedRevision: null, request,
+        appliedEventKeys: new Set(),
         requestedStepNodeIds: new Set(), recoveredStepNodes: new Map(),
       },
     });
@@ -747,14 +754,20 @@ export function createSessionRuntime({
       return stepNodesChanged;
     }
     const { assistantEvents } = latestConversationTurn(events || []);
-    assistantEvents.forEach((event) => appendEvent(live.message, event));
+    const newEvents = assistantEvents.filter((event) => {
+      const key = managedEventKey(event);
+      if (live.appliedEventKeys.has(key)) return false;
+      live.appliedEventKeys.add(key);
+      return true;
+    });
+    newEvents.forEach((event) => appendEvent(live.message, event));
     if (context) attachStepNodes(live.message.items, context);
     if (revision) live.hydratedRevision = revision;
     // Snapshot hydration is a recovery transaction, not a streamed update:
     // commit it synchronously so the recovered bubble never waits for a
     // throttle interval or another SSE token before becoming visible.
     renderTimeline(live.view, live.message, live.shownPlots);
-    return assistantEvents.length > 0;
+    return newEvents.length > 0;
   }
 
   function finishManagedPresentation(live) {
@@ -813,6 +826,9 @@ export function createSessionRuntime({
       }
       try {
         const event = JSON.parse(data);
+        const eventKey = managedEventKey(event);
+        if (live.appliedEventKeys.has(eventKey)) return;
+        live.appliedEventKeys.add(eventKey);
         updateManagedPhase(live, applyAssistantMessageEvent(live.message, event));
         void recoverManagedStepNodes(live, event);
       } catch (_) { /* malformed replay event */ }
diff --git a/web/vite-frontend/src/main.js b/web/vite-frontend/src/main.js
index e564eee..d016344 100644
--- a/web/vite-frontend/src/main.js
+++ b/web/vite-frontend/src/main.js
@@ -100,6 +100,7 @@ const state = {
   sessionSummaries: {},   // { sessionId: "summary text" }
   summaryGeneratedFor: new Set(),  // sessionIds that have triggered summary generation
   remoteJobs: [],
+  remoteJobGroups: [],
   appMode: "workspace",
   evaluationCatalog: [],
   evaluationCatalogTotal: null,
diff --git a/web/vite-frontend/src/styles/sessions.css b/web/vite-frontend/src/styles/sessions.css
index 0bed473..f6c04cf 100644
--- a/web/vite-frontend/src/styles/sessions.css
+++ b/web/vite-frontend/src/styles/sessions.css
@@ -386,6 +386,18 @@ body[data-theme="light"] .custom-select-options li[aria-selected="true"] {
   line-height: 1;
   text-transform: uppercase;
 }
+.remote-job-group {
+  display: block;
+  max-width: 100%;
+  margin-top: 3px;
+  overflow: hidden;
+  color: var(--accent);
+  font-size: var(--font-size-micro);
+  font-weight: var(--font-weight-semibold);
+  line-height: 1.2;
+  text-overflow: ellipsis;
+  white-space: nowrap;
+}
 .remote-job-id {
   display: block;
   min-width: 0;
diff --git a/web/vite-frontend/test/managedRunAttachment.test.js b/web/vite-frontend/test/managedRunAttachment.test.js
index c710d2a..517a4fa 100644
--- a/web/vite-frontend/test/managedRunAttachment.test.js
+++ b/web/vite-frontend/test/managedRunAttachment.test.js
@@ -203,6 +203,55 @@ test("the wakeup trigger folds in from durable history while the run is still st
   assert.ok(counters.historyLoads >= 1);
 });
 
+test("snapshot recovery does not reapply durable events when the session revision changes", async (t) => {
+  const { runtime, counters } = createHarness(t);
+  const events = [
+    { ...snapshotEvents[0], timestamp: 2 },
+    {
+      id: "call-1", author: "MatCreator",
+      content: { parts: [{ functionCall: { id: "tool-1", name: "first_task", args: {} } }] },
+    },
+    snapshotEvents[1],
+    {
+      id: "call-2", author: "MatCreator",
+      content: { parts: [{ functionCall: { id: "tool-2", name: "second_task", args: {} } }] },
+    },
+  ];
+  globalThis.fetch = async (url) => {
+    if (url === "/events") {
+      counters.streams += 1;
+      return new Response(new ReadableStream({ start() {} }));
+    }
+    counters.historyLoads += 1;
+    return Response.json({
+      events, state: { agent_mode: "normal" }, userId: "alice",
+      event_meta: events.map((_, index) => ({ index, cursor: `c${index}`, turn_id: "turn-1" })),
+      pagination: { start_index: 0, total_count: events.length },
+      revision: `r${counters.historyLoads}`,
+    });
+  };
+
+  await runtime.loadSession("session-1", "alice");
+  const attached = runtime.startManagedRunReconnect(
+    {
+      run_id: "wakeup-run", status: "running", created_at: 1,
+      earliest_sequence: 2,
+    },
+    "session-1",
+    "alice",
+  );
+  assert.ok(attached);
+  t.after(() => attached.controller.abort());
+
+  await runtime.loadSession("session-1", "alice");
+
+  const renderedText = attached.message.items
+    .filter((item) => item.type === "text")
+    .map((item) => item.text)
+    .join("|");
+  assert.equal(renderedText, "Job finished.");
+});
+
 test("a wakeup mounts alongside a presenting turn owned by another key without clearing it", async (t) => {
   const { runtime, state, counters, settle } = createHarness(t);
   // Owner-key normalization can leave the presenting request under another
diff --git a/web/vite-frontend/test/remoteJobs.test.js b/web/vite-frontend/test/remoteJobs.test.js
index 2f7a586..c089d79 100644
--- a/web/vite-frontend/test/remoteJobs.test.js
+++ b/web/vite-frontend/test/remoteJobs.test.js
@@ -199,6 +199,7 @@ function createFixture({ controllerOverrides = {}, skin = "rack-lab", windowOver
   const state = {
     activeSessionUserId: "owner-1",
     remoteJobs: [],
+    remoteJobGroups: [],
     sessionId: "session-1",
     userId: "owner-1",
   };
@@ -1519,3 +1520,60 @@ test("latest provider identity is visible and unsupported pause is capability-ga
   assert.equal(card.querySelector(".remote-job-action.terminate").disabled, false);
   fixture.controller.destroy();
 });
+
+test("grouped jobs show their group identity and policy metadata", async () => {
+  const data = {
+    jobs: [{
+      job_id: "mc-job-42",
+      group_id: "group-abc-123",
+      external_id: "bohr-batch-314",
+      provider: "bohr_batchjob",
+      status: "running",
+      snapshot: { provider_status: "RUNNING" },
+    }],
+    groups: [{
+      group_id: "group-abc-123",
+      name: "parameter-sweep",
+      expected_jobs: 2,
+      delivery_reason: null,
+    }],
+  };
+  const fixture = createFixture({
+    controllerOverrides: {
+      dummyMode: false,
+      httpClient: { getJson: async () => data },
+    },
+  });
+
+  await fixture.controller.load("session-1", "owner-1");
+
+  const card = fixture.list.children[0];
+  assert.equal(card.querySelector(".remote-job-group").textContent, "Group · parameter-sweep");
+  assert.equal(card.querySelector(".remote-job-group").title, "Group ID: group-abc-123");
+  const rowText = card.querySelectorAll(".remote-job-detail-row")
+    .map((row) => `${row.children[0]?.textContent}${row.children[1]?.textContent}`)
+    .join(" | ");
+  assert.match(rowText, /Group IDgroup-abc-123/);
+  assert.match(rowText, /Group nameparameter-sweep/);
+  assert.match(rowText, /Expected jobs2/);
+  assert.deepEqual(fixture.state.remoteJobGroups, data.groups);
+  fixture.controller.destroy();
+});
+
+test("ungrouped jobs do not show group metadata", () => {
+  const fixture = createFixture();
+  fixture.controller.setPresentationJobs([{
+    job_id: "mc-job-42",
+    external_id: "bohr-batch-314",
+    provider: "bohr_batchjob",
+    status: "running",
+    snapshot: { provider_status: "RUNNING" },
+  }]);
+
+  const card = fixture.list.children[0];
+  assert.equal(card.querySelector(".remote-job-group"), null);
+  const labels = card.querySelectorAll(".remote-job-detail-row")
+    .map((row) => row.children[0]?.textContent);
+  assert.equal(labels.includes("Group ID"), false);
+  fixture.controller.destroy();
+});