feat: extract portable runtime batch

README.md

@@ -33,6 +33,16 @@ Version `0.1.x` contains the converged foundation layer:
   parsing;
 - parallel tool execution and the complete agent-loop orchestration engine,
   with product behavior isolated behind typed hooks.
+- append-only, scope-isolated durable run journals with content-addressed blob
+  attachments, projections, and event-sourced notes;
+- MiniDAG execution, declarative pipeline models, dynamic graph construction,
+  and scoped service/agent/pipeline/topology registries;
+- reusable LLM middleware for proxying, summarization, skill injection, and
+  stream-repetition detection;
+- portable loop observers for budget/context/wall-clock guards, finalization,
+  repetition detection, SSE events, task boards, and bounded trajectories;
+- shared agent-bus stop signals/pools, filesystem skill loaders, session
+  history, renewable wall-time leases, and language helpers.
 
 The initial extraction is based on the already-merged integration branches:
 
@@ -43,9 +53,10 @@ Those revisions are provenance, not runtime dependencies. AgentCore tests and
 builds without either product checkout.
 
 The shared loop accepts product decisions through explicit hooks. Products own
-execution-context storage, endpoint registries, metering, deadline policy, and
-sandbox mount policy; they no longer need private copies of the orchestration,
-parsing, context-management, or spill-storage engines.
+composition roots, endpoint configuration, metering policy, durable-journal
+root/scope selection, and sandbox mount policy; they no longer need private
+copies of the orchestration, parsing, context-management, DAG, observer, or
+spill-storage engines.
 
 ## Repository boundary
 
@@ -57,9 +68,9 @@ Code belongs in AgentCore when it:
 - accepts product behavior through typed inputs or explicit hooks;
 - has tests that run without either product repository installed.
 
-Provider clients, environment/config loading, session affinity, checkpoint
-persistence, workflow assembly, sandbox mounting, and product-specific
-observers remain in their product.
+Provider clients, session affinity, user/session retention policy, checkpoint
+association, workflow node implementations, sandbox mounting/authorization,
+UI history, and product-specific observers remain in their product.
 
 ## Development
 
@@ -132,7 +143,14 @@ edit both products' core copies, that is evidence it belongs here.
    bounded LLM summary, tier selection, observability, and spill storage.
 6. **Runtime foundation** (complete in AgentCore): task identity/status,
    execution ContextVars, kernel events, event dispatch, and tool permissions.
-7. Remove product compatibility facades after downstream imports have moved to
+7. **Portable runtime batch** (complete in AgentCore): durable run journals,
+   MiniDAG/graph construction, middleware/observers, agent-bus coordination
+   primitives (shared pools and stop signals), skills, session history, models,
+   and scoped registries.
+8. **Agent-bus scheduling** (next): extract the shared communication models,
+   runtime, spawn guard, bus, and fan-in orchestration behind explicit product
+   hooks. Product tool assembly and workflow-specific policy remain downstream.
+9. Remove product compatibility facades after downstream imports have moved to
    `agent_core`.
 
 Each slice must leave product CI green and must not depend on a floating branch.

agent_core/components/__init__.py

@@ -0,0 +1 @@
+"""Portable agent runtime components."""

agent_core/components/agent_bus/__init__.py

@@ -0,0 +1 @@
+"""Portable agent coordination primitives."""

agent_core/components/agent_bus/shared_pool.py

@@ -0,0 +1,43 @@
+# pyright: reportUnknownVariableType=false, reportUnknownParameterType=false, reportUnknownArgumentType=false, reportUnknownMemberType=false, reportMissingTypeArgument=false, reportUnnecessaryIsInstance=false, reportUnusedFunction=false, reportAttributeAccessIssue=false, reportUnnecessaryComparison=false
+"""Shared artifact pool for parallel sub-agent execution.
+Holds dict-shaped artifacts produced concurrently by sub-agents and dedups
+by ``id`` field when present. Domain-specific subclasses live in their
+owning workflow — see its own ``shared_pool`` module.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+
+class SharedArtifactPool:
+    """Process-local artifact pool shared across parallel sub-agents.
+    Items are stored as shallow copies; if an item carries an ``id`` field,
+    later duplicates with the same id are silently dropped.
+    """
+
+    def __init__(self) -> None:
+        self._items: list[dict[str, Any]] = []
+        self._seen_ids: set[str] = set()
+        self._lock = asyncio.Lock()
+
+    async def add(self, items: list[dict[str, Any]]) -> None:
+        if not items:
+            return
+        async with self._lock:
+            for item in items:
+                if not isinstance(item, dict):
+                    continue
+                item_id = str(item.get("id", "")).strip()
+                if item_id and item_id in self._seen_ids:
+                    continue
+                if item_id:
+                    self._seen_ids.add(item_id)
+                self._items.append(dict(item))
+
+    def get_all(self) -> list[dict[str, Any]]:
+        return [dict(item) for item in self._items]
+
+    def count(self) -> int:
+        return len(self._items)

agent_core/components/agent_bus/stop_signal.py

@@ -0,0 +1,78 @@
+"""Process-wide cooperative stop-signal registry for sub-agents."""
+
+from __future__ import annotations
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class SubAgentStopRegistry:
+    """Pending cooperative-stop requests keyed by ``session_id``.
+    Each entry records the ``job_id`` the stop was requested against. Sessions
+    are reusable (``assign_task`` again after a stop), so a stop that outlives
+    its job — the job finished before the observer consumed it — must NOT roll
+    back a later task on the same session. :meth:`clear_stale` drops such
+    orphans at the next task's start.
+    Single-event-loop access, so a plain ``dict`` is race-free for our needs
+    (assignment / ``pop`` / membership are atomic w.r.t. the loop).
+    """
+
+    def __init__(self) -> None:
+        # session_id -> job_id the stop was requested against ("" if unknown).
+        self._requested: dict[str, str] = {}
+
+    def request_stop(self, session_id: str, job_id: str = "") -> None:
+        sid = (session_id or "").strip()
+        if not sid:
+            return
+        self._requested[sid] = job_id or ""
+        logger.info(
+            "Cooperative stop requested for sub-agent session=%s (job=%s)",
+            sid,
+            job_id or "?",
+        )
+
+    def consume(self, session_id: str) -> bool:
+        """Return True exactly once per request, clearing the flag.
+        Called by the observer at each turn boundary so the stop prompt is
+        injected only once even if more turns follow.
+        """
+        sid = (session_id or "").strip()
+        if sid in self._requested:
+            del self._requested[sid]
+            return True
+        return False
+
+    def clear_stale(self, session_id: str, current_job_id: str) -> bool:
+        """Drop a pending stop requested against a DIFFERENT job.
+        Called at the start of each new task on a (reusable) session so a stop
+        that outlived its job never fires on turn 1 of the next task. A stop
+        for the *current* job is preserved. Returns True if one was cleared.
+        """
+        sid = (session_id or "").strip()
+        pending = self._requested.get(sid)
+        if pending is not None and pending != (current_job_id or ""):
+            del self._requested[sid]
+            logger.info(
+                "Dropped stale stop for session=%s (queued for job=%s, now job=%s)",
+                sid,
+                pending or "?",
+                current_job_id or "?",
+            )
+            return True
+        return False
+
+    def is_requested(self, session_id: str) -> bool:
+        return (session_id or "").strip() in self._requested
+
+
+_REGISTRY = SubAgentStopRegistry()
+
+
+def get_stop_registry() -> SubAgentStopRegistry:
+    """Return the process-wide stop registry singleton."""
+    return _REGISTRY
+
+
+__all__ = ["SubAgentStopRegistry", "get_stop_registry"]

agent_core/components/finalization/__init__.py

@@ -0,0 +1 @@
+"""Portable final-answer recovery primitives."""