feat: training data pipeline, solver VL-JEPA, RPB deployment plan

- Rewrite PLAN.md to focus on RPB jurisdiction deployment (phases 1-4)
- Add training_attestor and training_feed modules for agent-trace training pipeline
- Update solver with VL-JEPA training support
- Extend node service with governance bridge initialization
- ATN runtime refinements: config, conversation export, session management
- Add tests for training attestor, training feed, and solver VL-JEPA integration
- Update governance integration tests

Author: EightRice

PLAN.md

@@ -22,6 +22,8 @@
 from enum import Enum
 from typing import Any, TYPE_CHECKING
 
+from .events import EventType
+
 if TYPE_CHECKING:
     from .config import AutonetConfig
     from .events import EventBus
@@ -85,7 +87,8 @@ class AutonetBridge:
     task so it doesn't block the ATN event loop.
     """
 
-    def __init__(self, config: AutonetConfig, event_bus: EventBus | None = None) -> None:
+    def __init__(self, config: AutonetConfig, event_bus: EventBus | None = None,
+                 data_dir: str = "") -> None:
         self.config = config
         self.state = AutonetState()
         self._events = event_bus
@@ -96,6 +99,8 @@ def __init__(self, config: AutonetConfig, event_bus: EventBus | None = None) ->
         self._published_standards_hash: str = ""
         self._published_tx_hash: str = ""
         self._user_contract_address: str = ""
+        # Training data feed (Story 3.2) — ATN data dir for JSONL files
+        self._data_dir = data_dir
 
         # Apply config values to state
         if config.wallet_address:
@@ -106,6 +111,13 @@ def __init__(self, config: AutonetConfig, event_bus: EventBus | None = None) ->
         if config.chain_id:
             self.state.chain_id = config.chain_id
 
+        # Subscribe to execution events for training data feed
+        if self._events:
+            self._events.subscribe(
+                EventType.EXECUTION_COMPLETED,
+                self._on_execution_completed,
+            )
+
     async def _emit(self, event_type_name: str, data: dict[str, Any] | None = None) -> None:
         """Emit an event if the event bus is available."""
         if not self._events:
@@ -119,6 +131,20 @@ async def _emit(self, event_type_name: str, data: dict[str, Any] | None = None)
                 data=data or self.state.to_dict(),
             ))
 
+    async def _on_execution_completed(self, event) -> None:
+        """Handle EXECUTION_COMPLETED events from the EventBus.
+
+        Story 3.2: When an agent execution completes, notify the training
+        service so it knows new training data is available on disk.
+        """
+        data = event.data or {}
+        agent_id = data.get("agent_id", "")
+        execution_id = data.get("execution_id", "")
+        status = data.get("status", "")
+
+        if self._service:
+            self._service.notify_execution(agent_id, execution_id, status)
+
     async def start(self) -> dict[str, Any]:
         """Start the autonet training service.
 
@@ -144,10 +170,12 @@ async def start(self) -> dict[str, Any]:
                 self._autonet_config.blockchain.rpc_url = self.config.rpc_url
             if self.config.chain_id:
                 self._autonet_config.blockchain.chain_id = self.config.chain_id
+            if self.config.private_key:
+                self._autonet_config.blockchain.private_key = self.config.private_key
 
             # Create and start the service in a background thread
             # (AutonetService.start() is blocking)
-            self._service = AutonetService(self._autonet_config)
+            self._service = AutonetService(self._autonet_config, data_dir=self._data_dir)
             self._task = asyncio.create_task(self._run_service())
 
             self.state.status = AutonetStatus.RUNNING

atn/autonet_service.py

@@ -106,6 +106,7 @@ class AutonetConfig:
     # Blockchain connection (overrides autonet.yaml if set)
     rpc_url: str = ""                   # e.g. "https://node.shadownet.etherlink.com"
     chain_id: int = 0                   # e.g. 127823 for Etherlink Shadownet
+    private_key: str = ""               # Hex private key for signing attestation txns
     # Wallet is managed externally (MetaMask etc.) — we just track the address
     wallet_address: str = ""            # Connected wallet address (empty = not connected)
 
@@ -267,6 +268,7 @@ def load_config(path: Path | None = None) -> ATNConfig:
             config_path=resolved.get("config_path", ""),
             rpc_url=resolved.get("rpc_url", ""),
             chain_id=resolved.get("chain_id", 0),
+            private_key=resolved.get("private_key", ""),
             wallet_address=resolved.get("wallet_address", ""),
         )
 

atn/config.py

@@ -120,20 +120,27 @@ def get_turns_page(self, limit: int = 50, offset: int | None = None) -> tuple[li
             return [], total
         return list(self._turns[start:end]), total
 
-    def get_history_for_prompt(self) -> str:
+    def get_history_for_prompt(self, exclude_last: int = 0) -> str:
         """Build a conversation history string that fits within the token budget.
 
         Returns a formatted string of previous turns, newest last, trimmed
         from the front (oldest) if the total exceeds the budget.  Returns
         empty string if no history.
+
+        Args:
+            exclude_last: Number of turns to exclude from the end (e.g. 1 to
+                skip the current user message that was just recorded).
         """
-        if not self._turns:
+        turns = self._turns
+        if exclude_last > 0:
+            turns = turns[:-exclude_last] if len(turns) > exclude_last else []
+        if not turns:
             return ""
 
         # Build formatted turns
         formatted: list[str] = []
         _ROLE_PREFIX = {"user": "User", "assistant": "Orchestrator", "system": "System"}
-        for turn in self._turns:
+        for turn in turns:
             prefix = _ROLE_PREFIX.get(turn.role, turn.role.title())
             formatted.append(f"{prefix}: {turn.content}")
 

atn/conversation.py

@@ -1186,11 +1186,11 @@ async def _post_message(runtime: Runtime, input: dict[str, Any]) -> dict[str, An
     priority = MessagePriority(input.get("priority", "normal"))
     instruction = (input.get("data") or {}).get("instruction", "")
 
-    # For cognitive agents with an instruction, persist the message in
-    # the agent's conversation store so it appears in its chat thread.
-    if defn.mode == AgentMode.COGNITIVE and instruction:
-        store = runtime.get_agent_conversation_store(target)
-        store.add_user_turn(instruction)
+    # NOTE: Do NOT add_user_turn here.  The execution engine records the
+    # user turn right before calling the LLM (execution_engine.py lines
+    # 624-636), which is the single place responsible for persisting turns
+    # for both the orchestrator and child agents.  Adding it here caused
+    # duplicate user messages in the conversation store.
 
     msg = InboxMessage(
         id=InboxMessage.generate_id(),

atn/orchestrator/tools.py

@@ -194,9 +194,13 @@ def __init__(self, event_bus: EventBus, data_dir: Path | None = None, config: AT
         # Voice service (lazy)
         self.voice = None  # type: Any
 
-        # Autonet service
+        # Autonet service (Story 3.2: pass data_dir for training data feed)
         from ..autonet_service import AutonetBridge
-        self.autonet = AutonetBridge(self._config.autonet, event_bus=self.events)
+        self.autonet = AutonetBridge(
+            self._config.autonet,
+            event_bus=self.events,
+            data_dir=str(self._config.data_dir),
+        )
 
         # Tool registry
         from ..tool_registry import ToolRegistry

atn/runtime/__init__.py

@@ -613,19 +613,23 @@ async def _execute_cognitive_agent(
             # --- Session resume / history ---
             session_id = ""
             if is_orchestrator:
-                # Build history from PRIOR turns before recording this one.
                 session_id = getattr(sub_provider, '_session_id', "") or ""
-                if not session_id:
-                    history = self.session_manager.conversation.get_history_for_prompt()
-                else:
-                    history = ""
-                # Store the raw user message (skip if already recorded by
-                # ws_server or send_agent_message to avoid duplicates).
+                # Record the user message (skip if already recorded by
+                # ws_server to avoid duplicates in the conversation store).
                 existing_turns = self.session_manager.conversation.get_turns()
                 if (not existing_turns
                         or existing_turns[-1].role != "user"
                         or existing_turns[-1].content != user_message):
                     self.session_manager.conversation.add_user_turn(user_message)
+                # Build prior-turn history for non-session providers,
+                # excluding the current user message (exclude_last=1) so
+                # it doesn't appear in both history AND the prompt.
+                if not session_id:
+                    history = self.session_manager.conversation.get_history_for_prompt(
+                        exclude_last=1,
+                    )
+                else:
+                    history = ""
                 # Prepend conversation history for non-session providers
                 if history:
                     user_message = history + "\n\nUser: " + user_message

atn/runtime/execution_engine.py

@@ -86,17 +86,21 @@ async def send_agent_message(self, agent_id: str, text: str) -> dict:
             return {"error": f"Agent '{agent_id}' is not a cognitive agent"}
 
         store = self.get_agent_conversation_store(agent_id)
-        store.add_user_turn(text)
 
         # Mid-session injection — only if actually running
         is_running = self.registry._running_count.get(agent_id, 0) > 0
         if is_running:
             provider = self.provider_manager._active_providers.get(agent_id)
             if provider is not None:
+                # Record in conversation store here — the execution engine
+                # won't see this message (it goes directly to the provider).
+                store.add_user_turn(text)
                 await provider.send_user_message(text)
                 return {"status": "injected", "agent_id": agent_id}
 
-        # Not running — post to inbox
+        # Not running — post to inbox.  Do NOT add_user_turn here;
+        # the execution engine records it when it drains the inbox
+        # (avoiding duplicates in the conversation store).
         msg = InboxMessage(
             id=InboxMessage.generate_id(),
             type=MessageType.WORK,

atn/runtime/session_manager.py

@@ -654,7 +654,10 @@ async def _do_oauth() -> None:
         if msg_type == "post_message" and "source" not in args:
             args["source"] = "user"
 
-        # Record user turn in conversation history when messaging the orchestrator
+        # Record user turn in conversation history early so the UI can fetch
+        # it immediately via get_conversation (the frontend does an optimistic
+        # add, but a history reload would lose it without this).
+        # The execution engine skips re-adding it (dedup check at line ~624).
         if (msg_type == "post_message"
                 and args.get("target") == "orchestrator"
                 and args.get("source") == "user"):