Add ultrawork logs command and acceptance test

New UltraworkStore.logs() method reads log tail with configurable max bytes
New ultrawork logs <worker_id> [--bytes N] CLI subcommand
New acceptance test covers start → list → show → logs → stop lifecycle
Wire up handler exports and parser registration; update acceptance docs

Author: johnteee

docs/acceptance.md

@@ -11,9 +11,9 @@ TeaAgent acceptance tests live under `tests/acceptance/` and verify user-facing
 - 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.
+- Long-running worker flow: background worker start, list, show, log tail, and stop lifecycle.
 
 ## Next User Stories
 
 - Live provider conformance gated by environment variables.
-- 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/cli/__init__.py

@@ -39,6 +39,7 @@
     model_smoke,
     start_tui,
     ultrawork_list_command,
+    ultrawork_logs_command,
     ultrawork_show_command,
     ultrawork_start_command,
     ultrawork_stop_command,
@@ -118,6 +119,7 @@ def build_parser() -> argparse.ArgumentParser:
             'graphqlite_migrate': graphqlite_migrate,
             'ultrawork_start': ultrawork_start_command,
             'ultrawork_list': ultrawork_list_command,
+            'ultrawork_logs': ultrawork_logs_command,
             'ultrawork_show': ultrawork_show_command,
             'ultrawork_stop': ultrawork_stop_command,
             'workspace_tools': workspace_tools_metadata,

teaagent/cli/_handlers/__init__.py

@@ -35,6 +35,7 @@
     graphqlite_smoke,
     start_tui,
     ultrawork_list_command,
+    ultrawork_logs_command,
     ultrawork_show_command,
     ultrawork_start_command,
     ultrawork_stop_command,
@@ -76,6 +77,7 @@
     'model_smoke',
     'start_tui',
     'ultrawork_list_command',
+    'ultrawork_logs_command',
     'ultrawork_show_command',
     'ultrawork_start_command',
     'ultrawork_stop_command',

teaagent/cli/_handlers/_misc.py

@@ -100,6 +100,15 @@ def ultrawork_show_command(args: argparse.Namespace) -> int:
     return 0
 
 
+def ultrawork_logs_command(args: argparse.Namespace) -> int:
+    try:
+        print_json(UltraworkStore(args.root).logs(args.worker_id, max_bytes=args.bytes))
+    except FileNotFoundError as exc:
+        print_json({'status': 'error', 'message': str(exc)})
+        return 1
+    return 0
+
+
 def ultrawork_stop_command(args: argparse.Namespace) -> int:
     try:
         print_json(UltraworkStore(args.root).stop(args.worker_id))

teaagent/cli/_misc_parsers.py

@@ -40,6 +40,7 @@ def register(
         handlers['ultrawork_start'],
         handlers['ultrawork_list'],
         handlers['ultrawork_show'],
+        handlers['ultrawork_logs'],
         handlers['ultrawork_stop'],
     )
     _workspace(subparsers, handlers['workspace_tools'], handlers['workspace_openapi'])
@@ -268,6 +269,7 @@ def _ultrawork(
     start_handler: Callable,
     list_handler: Callable,
     show_handler: Callable,
+    logs_handler: Callable,
     stop_handler: Callable,
 ) -> None:
     ultrawork = subparsers.add_parser(
@@ -308,6 +310,14 @@ def _ultrawork(
     show.add_argument('--root', default='.', help='Workspace root.')
     show.set_defaults(func=show_handler)
 
+    logs = subs.add_parser('logs', help='Show one worker log tail.')
+    logs.add_argument('worker_id', help='Worker id to inspect.')
+    logs.add_argument('--root', default='.', help='Workspace root.')
+    logs.add_argument(
+        '--bytes', type=int, default=64_000, help='Maximum log bytes to return.'
+    )
+    logs.set_defaults(func=logs_handler)
+
     stop = subs.add_parser('stop', help='Stop a running worker.')
     stop.add_argument('worker_id', help='Worker id to stop.')
     stop.add_argument('--root', default='.', help='Workspace root.')

teaagent/ultrawork.py

@@ -102,6 +102,24 @@ def show(self, worker_id: str) -> dict[str, Any]:
         data['alive'] = self._is_alive(int(data['pid']))
         return data
 
+    def logs(self, worker_id: str, *, max_bytes: int = 64_000) -> dict[str, Any]:
+        data = self.show(worker_id)
+        log_path = Path(str(data['log_path']))
+        if not log_path.exists():
+            content = ''
+        else:
+            with log_path.open('rb') as fh:
+                if max_bytes > 0:
+                    fh.seek(0, os.SEEK_END)
+                    size = fh.tell()
+                    fh.seek(max(0, size - max_bytes), os.SEEK_SET)
+                content = fh.read().decode('utf-8', errors='replace')
+        return {
+            'worker_id': data['worker_id'],
+            'log_path': data['log_path'],
+            'content': content,
+        }
+
     def stop(self, worker_id: str, *, timeout_seconds: float = 2.0) -> dict[str, Any]:
         data = self.show(worker_id)
         pid = int(data['pid'])

tests/acceptance/test_ultrawork_flow.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import io
+import json
+import sys
+import tempfile
+import time
+import unittest
+from contextlib import redirect_stdout
+
+from teaagent import UltraworkStore
+from teaagent.cli import main
+
+
+class UltraworkFlowAcceptanceTests(unittest.TestCase):
+    def test_ultrawork_start_list_show_logs_and_stop_flow(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = UltraworkStore(tmp)
+            record = store.start(
+                [
+                    sys.executable,
+                    '-c',
+                    'import time; print("worker ready", flush=True); time.sleep(5)',
+                ],
+                label='acceptance-worker',
+            )
+            time.sleep(0.2)
+
+            list_out = io.StringIO()
+            with redirect_stdout(list_out):
+                list_code = main(['ultrawork', 'list', '--root', tmp])
+
+            show_out = io.StringIO()
+            with redirect_stdout(show_out):
+                show_code = main(['ultrawork', 'show', record.worker_id, '--root', tmp])
+
+            logs_out = io.StringIO()
+            with redirect_stdout(logs_out):
+                logs_code = main(['ultrawork', 'logs', record.worker_id, '--root', tmp])
+
+            stop_out = io.StringIO()
+            with redirect_stdout(stop_out):
+                stop_code = main(['ultrawork', 'stop', record.worker_id, '--root', tmp])
+
+            list_payload = json.loads(list_out.getvalue())
+            show_payload = json.loads(show_out.getvalue())
+            logs_payload = json.loads(logs_out.getvalue())
+            stop_payload = json.loads(stop_out.getvalue())
+
+            self.assertEqual(list_code, 0)
+            self.assertEqual(show_code, 0)
+            self.assertEqual(logs_code, 0)
+            self.assertEqual(stop_code, 0)
+            self.assertEqual(list_payload[0]['worker_id'], record.worker_id)
+            self.assertEqual(show_payload['label'], 'acceptance-worker')
+            self.assertTrue(show_payload['alive'])
+            self.assertIn('worker ready', logs_payload['content'])
+            self.assertFalse(stop_payload['alive'])
+            self.assertIn(stop_payload['stop_signal'], {'SIGTERM', 'SIGKILL'})
+
+
+if __name__ == '__main__':
+    unittest.main()