Add managed runtime context helper and acceptance test

New managed_runtime_context() helper builds tool metadata and workspace context for managed agent runs
ManagedAgentRunner.run() now captures context_keys and tool_count in audit log events and result metadata
New acceptance test covers context construction, forwarding, audit persistence, and result trace metadata
Export managed_runtime_context from package; update acceptance docs

Author: johnteee

docs/acceptance.md

@@ -10,10 +10,10 @@ TeaAgent acceptance tests live under `tests/acceptance/` and verify user-facing
 - TUI prompt approval workflow: approval prompt, destructive write, final run payload, and audit summary.
 - MCP client compatibility flow: bearer auth, session lifecycle, `tools/list`, `tools/call`, and session close.
 - A2A federation flow: remote discovery, partial endpoint failure, capability routing, delegation, context forwarding, and agent trace metadata.
+- Managed runtime flow: tool metadata context construction, workspace/request context forwarding, persisted managed task audit events, and result trace metadata.
 
 ## Next User Stories
 
 - Live provider conformance gated by environment variables.
-- Managed runtime flow that forwards task context/tools and records managed runtime audit events.
 - Long-running worker flow covering `ultrawork start`, heartbeat/status, logs, and stop.
 - Workspace edit flow covering hash reads, patch application, git status, test execution, and final diff summary.

teaagent/__init__.py

@@ -64,6 +64,7 @@
     ModelConformanceResult,
     run_model_conformance,
 )
+from teaagent.managed_runtime import managed_runtime_context
 from teaagent.mcp_client import MCPClientError, MCPHTTPClient
 from teaagent.mcp_http import build_mcp_http_server, serve_mcp_http
 from teaagent.mcp_server import handle_mcp_request, serve_mcp_stdio
@@ -252,6 +253,7 @@
     'graph_retrieve',
     'handle_mcp_request',
     'handle_stateless_tool_request',
+    'managed_runtime_context',
     'parse_model_decision',
     'parse_permission_mode',
     'preflight',

teaagent/managed_runtime.py

@@ -21,6 +21,19 @@ class ManagedRunResult:
     metadata: dict[str, Any] = field(default_factory=dict)
 
 
+def managed_runtime_context(
+    registry: Any,
+    *,
+    workspace_root: Optional[str] = None,
+    extra: Optional[dict[str, Any]] = None,
+) -> dict[str, Any]:
+    context = dict(extra or {})
+    context['tools'] = registry.mcp_metadata()
+    if workspace_root is not None:
+        context['workspace_root'] = workspace_root
+    return context
+
+
 class ManagedAgentRunner:
     def __init__(
         self, adapter: ManagedRuntimeAdapter, *, runtime_name: str = ''
@@ -37,10 +50,18 @@ def run(
         run_id: str = '',
     ) -> ManagedRunResult:
         ctx = context or {}
+        context_keys = sorted(ctx.keys())
+        tools = ctx.get('tools', [])
+        tool_count = len(tools) if isinstance(tools, list) else 0
         _log = None if audit_logger is _AUDIT_UNSET else audit_logger
         if _log is not None:
             _log.record(
-                'managed_task_started', run_id, runtime=self._runtime_name, task=task
+                'managed_task_started',
+                run_id,
+                runtime=self._runtime_name,
+                task=task,
+                context_keys=context_keys,
+                tool_count=tool_count,
             )
         try:
             output = self._adapter.run_task(task, context=ctx)
@@ -51,6 +72,8 @@ def run(
                     run_id,
                     runtime=self._runtime_name,
                     error=str(exc),
+                    context_keys=context_keys,
+                    tool_count=tool_count,
                 )
             raise
         if _log is not None:
@@ -59,8 +82,18 @@ def run(
                 run_id,
                 runtime=self._runtime_name,
                 output_length=len(output),
+                context_keys=context_keys,
+                tool_count=tool_count,
             )
-        return ManagedRunResult(output=output, runtime=self._runtime_name)
+        return ManagedRunResult(
+            output=output,
+            runtime=self._runtime_name,
+            metadata={
+                'run_id': run_id,
+                'context_keys': context_keys,
+                'tool_count': tool_count,
+            },
+        )
 
     def healthy(self) -> bool:
         return self._adapter.health_check()

tests/acceptance/test_managed_runtime_flow.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from teaagent.audit import AuditLogger
+from teaagent.managed_runtime import ManagedAgentRunner, managed_runtime_context
+from teaagent.tools import ToolAnnotations, ToolRegistry
+
+
+class _CapturingManagedRuntime:
+    def __init__(self) -> None:
+        self.received_task = ''
+        self.received_context: dict = {}
+
+    def run_task(self, task: str, *, context: dict) -> str:
+        self.received_task = task
+        self.received_context = context
+        return f'managed:{len(context["tools"])}'
+
+    def health_check(self) -> bool:
+        return True
+
+
+class ManagedRuntimeFlowAcceptanceTests(unittest.TestCase):
+    def test_managed_runtime_receives_tool_context_and_persists_audit(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            audit_path = Path(tmp) / 'managed-run.jsonl'
+            audit = AuditLogger(audit_path)
+            registry = ToolRegistry()
+            registry.register(
+                name='workspace_read_file',
+                description='Read a workspace file',
+                input_schema={'type': 'object'},
+                output_schema={'type': 'object'},
+                annotations=ToolAnnotations(read_only=True),
+                handler=lambda _args: {'content': 'ok'},
+            )
+            runtime = _CapturingManagedRuntime()
+            runner = ManagedAgentRunner(runtime, runtime_name='acceptance-runtime')
+
+            context = managed_runtime_context(
+                registry, workspace_root=tmp, extra={'request_id': 'req-1'}
+            )
+            result = runner.run(
+                'summarize workspace',
+                context=context,
+                audit_logger=audit,
+                run_id='managed-1',
+            )
+            events = [
+                json.loads(line)
+                for line in audit_path.read_text(encoding='utf-8').splitlines()
+            ]
+
+            self.assertEqual(result.output, 'managed:1')
+            self.assertEqual(result.runtime, 'acceptance-runtime')
+            self.assertEqual(result.metadata['run_id'], 'managed-1')
+            self.assertEqual(result.metadata['tool_count'], 1)
+            self.assertIn('tools', result.metadata['context_keys'])
+            self.assertEqual(runtime.received_task, 'summarize workspace')
+            self.assertEqual(runtime.received_context['request_id'], 'req-1')
+            self.assertEqual(runtime.received_context['workspace_root'], tmp)
+            self.assertEqual(
+                runtime.received_context['tools'][0]['name'], 'workspace_read_file'
+            )
+            self.assertEqual(
+                [event['event_type'] for event in events],
+                ['managed_task_started', 'managed_task_completed'],
+            )
+            self.assertEqual(events[0]['payload']['tool_count'], 1)
+            self.assertEqual(events[1]['payload']['output_length'], len('managed:1'))
+
+
+if __name__ == '__main__':
+    unittest.main()