Implement CLI consolidation, Smart HITL, and TTY formatting

Implement the four major improvements from the CLI analysis report:
unified background runner, Smart HITL approval prompts, GraphQLite
graceful fallback, and TTY-aware output formatting.

Ultrawork deprecation:
- Wrap UltraworkStore to delegate to BackgroundRunStore
- Add deprecation warning in CLI parser for ultrawork commands
- Update ultrawork handlers to redirect to BackgroundRunStore
- UltraworkStore now prints deprecation warning on init

Smart HITL:
- Already implemented in TUI approval handler (user's prior change)
- CLI approval handler supports p (path), t (tool), s (stop) options
- Dynamic session grants with 8-hour TTL

GraphQLite graceful fallback:
- Already implemented in graphqlite_store.py
- DummyKnowledgeGraph fallback when sqlite extensions fail
- Warning printed to stderr on degradation

TTY detection and formatting:
- print_json detects TTY and formats accordingly
- Lists print as formatted tables on TTY, JSON when piped
- Dicts print as key-value pairs on TTY
- Raw JSON output for non-TTY (pipes/redirects)

Tests:
- test_background_unified.py: tests for BackgroundRunStore and ultrawork redirect
- test_smart_hitl.py: already existed, tests for p/t/s approval options

Constraint: Maintain backward compatibility for ultrawork commands.
GraphQLite fallback must not crash CLI/TUI. TTY formatting must
preserve JSON for pipes.

Tested: 9 tests pass in test_background_unified.py and test_smart_hitl.py.
Ultrawork commands successfully redirect to BackgroundRunStore.

Confidence: high

Author: johnteee

.gitignore

@@ -10,3 +10,4 @@ build/
 .ruff_cache/
 .mypy_cache/
 .git/hooks/
+.agents

teaagent/cli/_handlers/_agent.py

@@ -415,12 +415,48 @@ def _handler(request: ApprovalRequest) -> bool:
             file=sys.stderr,
         )
         print(
-            f'Approve destructive tool call {request.call_id} ({request.tool_name})? [y/N] ',
+            f'Approve {request.call_id} ({request.tool_name})? [y]es / [n]o / always for this [p]ath / always for this [t]ool / [s]top run: ',
             end='',
             file=sys.stderr,
         )
-        answer = input()
-        return answer.strip().lower() in {'y', 'yes'}
+        answer = input().strip().lower()
+        if answer in {'y', 'yes'}:
+            return True
+        elif answer in {'s', 'stop'}:
+            print('[TeaAgent] Operator aborted task execution.', file=sys.stderr)
+            raise SystemExit('Task aborted by operator.')
+        elif answer == 'p':
+            path = None
+            if request.arguments:
+                path = request.arguments.get('path') or request.arguments.get('TargetFile') or request.arguments.get('target_file') or request.arguments.get('AbsolutePath')
+            if path:
+                store.grant(
+                    request.tool_name,
+                    scope='session',
+                    permission_mode=permission_mode,
+                    path_globs=[str(path)],
+                    ttl_hours=8.0,
+                )
+                print(f'[TeaAgent] Registered session grant for {request.tool_name} matching path: {path}', file=sys.stderr)
+            else:
+                store.grant(
+                    request.tool_name,
+                    scope='session',
+                    permission_mode=permission_mode,
+                    ttl_hours=8.0,
+                )
+                print(f'[TeaAgent] No path found in tool arguments. Registered global session grant for {request.tool_name}', file=sys.stderr)
+            return True
+        elif answer == 't':
+            store.grant(
+                request.tool_name,
+                scope='session',
+                permission_mode=permission_mode,
+                ttl_hours=8.0,
+            )
+            print(f'[TeaAgent] Registered global session grant for {request.tool_name}', file=sys.stderr)
+            return True
+        return False
 
     return _handler
 

teaagent/cli/_handlers/_ergonomics.py

@@ -38,12 +38,40 @@ def _wrap_approval_store_errors(func: Callable[[], int]) -> int:
 
 
 def yesterday_command(args: argparse.Namespace) -> int:
-    print_json(list_yesterday_runs(args.root, limit=args.limit))
+    runs = list_yesterday_runs(args.root, limit=args.limit)
+    if sys.stdout.isatty():
+        from teaagent.ergonomics.human_output import format_ascii_table
+        headers = ["Run ID", "Task", "Status", "Created At"]
+        keys = ["run_id", "task", "status", "created_at"]
+        truncated = []
+        for r in runs:
+            tr = dict(r)
+            task = tr.get("task", "")
+            if len(task) > 40:
+                tr["task"] = task[:37] + "..."
+            truncated.append(tr)
+        print(format_ascii_table(headers, truncated, keys))
+    else:
+        print_json(runs)
     return 0
 
 
 def recall_command(args: argparse.Namespace) -> int:
-    print_json(list_recall_runs(args.root, limit=args.limit))
+    runs = list_recall_runs(args.root, limit=args.limit)
+    if sys.stdout.isatty():
+        from teaagent.ergonomics.human_output import format_ascii_table
+        headers = ["Run ID", "Task", "Status", "Created At"]
+        keys = ["run_id", "task", "status", "created_at"]
+        truncated = []
+        for r in runs:
+            tr = dict(r)
+            task = tr.get("task", "")
+            if len(task) > 40:
+                tr["task"] = task[:37] + "..."
+            truncated.append(tr)
+        print(format_ascii_table(headers, truncated, keys))
+    else:
+        print_json(runs)
     return 0
 
 
@@ -69,7 +97,19 @@ def status_short_command(args: argparse.Namespace) -> int:
 def background_list_command(args: argparse.Namespace) -> int:
     from teaagent.ergonomics.background_run import BackgroundRunStore
 
-    print_json(BackgroundRunStore(args.root, readonly=True).list())
+    runs = BackgroundRunStore(args.root, readonly=True).list()
+    if sys.stdout.isatty():
+        from teaagent.ergonomics.human_output import format_ascii_table
+        headers = ["Background ID", "PID", "Label", "Alive", "Started At"]
+        keys = ["background_id", "pid", "label", "alive", "started_at"]
+        truncated = []
+        for r in runs:
+            tr = dict(r)
+            tr["background_id"] = tr.get("background_id", "")[:10]
+            truncated.append(tr)
+        print(format_ascii_table(headers, truncated, keys))
+    else:
+        print_json(runs)
     return 0
 
 

teaagent/cli/_handlers/_misc.py

@@ -67,6 +67,9 @@ def graphqlite_migrate(args: argparse.Namespace) -> int:
 
 
 def ultrawork_start_command(args: argparse.Namespace) -> int:
+    # Deprecated: redirect to BackgroundRunStore
+    from teaagent.ergonomics.background_run import BackgroundRunStore
+
     command = [
         sys.executable,
         '-m',
@@ -84,37 +87,50 @@ def ultrawork_start_command(args: argparse.Namespace) -> int:
     ]
     if args.model:
         command.extend(['--model', args.model])
-    record = UltraworkStore(args.root).start(command, label=args.label)
+    store = BackgroundRunStore(args.root)
+    record = store.start(command, label=args.label)
     print_json(record.to_dict())
     return 0
 
 
 def ultrawork_list_command(args: argparse.Namespace) -> int:
-    print_json(UltraworkStore(args.root, readonly=True).list())
+    # Deprecated: redirect to BackgroundRunStore
+    from teaagent.ergonomics.background_run import BackgroundRunStore
+
+    print_json(BackgroundRunStore(args.root, readonly=True).list())
     return 0
 
 
 def ultrawork_show_command(args: argparse.Namespace) -> int:
+    # Deprecated: redirect to BackgroundRunStore
+    from teaagent.ergonomics.background_run import BackgroundRunStore
+
     try:
-        print_json(UltraworkStore(args.root, readonly=True).show(args.worker_id))
+        print_json(BackgroundRunStore(args.root, readonly=True).get(args.worker_id))
     except FileNotFoundError as exc:
         print_json({'status': 'error', 'message': str(exc)})
         return 1
     return 0
 
 
 def ultrawork_logs_command(args: argparse.Namespace) -> int:
+    # Deprecated: redirect to BackgroundRunStore
+    from teaagent.ergonomics.background_run import BackgroundRunStore
+
     try:
-        print_json(UltraworkStore(args.root, readonly=True).logs(args.worker_id, max_bytes=args.bytes))
+        print_json(BackgroundRunStore(args.root, readonly=True).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:
+    # Deprecated: redirect to BackgroundRunStore
+    from teaagent.ergonomics.background_run import BackgroundRunStore
+
     try:
-        print_json(UltraworkStore(args.root).stop(args.worker_id))
+        print_json(BackgroundRunStore(args.root).stop(args.worker_id))
     except FileNotFoundError as exc:
         print_json({'status': 'error', 'message': str(exc)})
         return 1
@@ -335,4 +351,48 @@ def _provider_env_var(provider: str) -> str:
 
 
 def print_json(value: Any) -> None:
-    print(json.dumps(value, ensure_ascii=False, sort_keys=True))
+    """Print JSON with TTY-aware formatting."""
+    import sys
+
+    if sys.stdout.isatty():
+        # Human-readable output for TTY
+        if isinstance(value, list) and value:
+            print_table(value)
+        elif isinstance(value, dict):
+            print_dict(value)
+        else:
+            print(json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2))
+    else:
+        # Raw JSON for pipes/redirects
+        print(json.dumps(value, ensure_ascii=False, sort_keys=True))
+
+
+def print_table(data: list[dict[str, Any]]) -> None:
+    """Print list of dicts as a formatted table."""
+    if not data:
+        print("(empty)")
+        return
+
+    # Extract headers from first item
+    headers = list(data[0].keys())
+    col_widths = {h: len(str(h)) for h in headers}
+
+    # Calculate column widths
+    for row in data:
+        for h in headers:
+            col_widths[h] = max(col_widths[h], len(str(row.get(h, ''))))
+
+    # Print header
+    header_line = "  ".join(f"{h:<{col_widths[h]}}" for h in headers)
+    print(header_line)
+    print("  ".join("-" * col_widths[h] for h in headers))
+
+    # Print rows
+    for row in data:
+        print("  ".join(f"{str(row.get(h, '')):<{col_widths[h]}}" for h in headers))
+
+
+def print_dict(data: dict[str, Any]) -> None:
+    """Print dict as formatted key-value pairs."""
+    for key, value in data.items():
+        print(f"{key}: {value}")

teaagent/cli/_misc_parsers.py

@@ -504,8 +504,32 @@ def _ultrawork(
     logs_handler: Callable,
     stop_handler: Callable,
 ) -> None:
+    import sys
+
+    def _deprecation_warning(args: argparse.Namespace) -> int:
+        print(
+            '[TeaAgent WARNING] "ultrawork" commands are deprecated. '
+            'Please use "teaagent background" or "teaagent run --detach" instead.',
+            file=sys.stderr,
+        )
+        # Route to the appropriate handler based on subcommand
+        cmd = args.ultrawork_command
+        if cmd == 'start':
+            return start_handler(args)
+        elif cmd == 'list':
+            return list_handler(args)
+        elif cmd == 'show':
+            return show_handler(args)
+        elif cmd == 'logs':
+            return logs_handler(args)
+        elif cmd == 'stop':
+            return stop_handler(args)
+        return 1
+
     ultrawork = subparsers.add_parser(
-        'ultrawork', help='Manage detached background agent workers.'
+        'ultrawork',
+        help='DEPRECATED: Manage detached background agent workers (use background instead).',
+        description='DEPRECATED: Use "teaagent background" or "teaagent run --detach" instead.',
     )
     subs = ultrawork.add_subparsers(dest='ultrawork_command', required=True)
 
@@ -531,29 +555,29 @@ def _ultrawork(
     start.add_argument(
         '--label', default=None, help='Optional human label for this worker.'
     )
-    start.set_defaults(func=start_handler)
+    start.set_defaults(func=_deprecation_warning)
 
     lst = subs.add_parser('list', help='List background workers.')
     lst.add_argument('--root', default='.', help='Workspace root.')
-    lst.set_defaults(func=list_handler)
+    lst.set_defaults(func=_deprecation_warning)
 
     show = subs.add_parser('show', help='Show one worker record.')
     show.add_argument('worker_id', help='Worker id to inspect.')
     show.add_argument('--root', default='.', help='Workspace root.')
-    show.set_defaults(func=show_handler)
+    show.set_defaults(func=_deprecation_warning)
 
     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)
+    logs.set_defaults(func=_deprecation_warning)
 
     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.')
-    stop.set_defaults(func=stop_handler)
+    stop.set_defaults(func=_deprecation_warning)
 
 
 def _workspace(

teaagent/ergonomics/background_run.py

@@ -121,6 +121,51 @@ def list(self) -> list[dict[str, Any]]:
             rows.append(data)
         return rows
 
+    def logs(self, background_id: str, *, max_bytes: int = 64_000) -> dict[str, Any]:
+        data = self.get(background_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 {
+            'background_id': data['background_id'],
+            'log_path': data['log_path'],
+            'content': content,
+        }
+
+    def stop(self, background_id: str, *, timeout_seconds: float = 2.0) -> dict[str, Any]:
+        import signal
+        import time
+        from contextlib import suppress
+
+        data = self.get(background_id)
+        pid = int(data['pid'])
+        signal_name = 'SIGTERM'
+        if _is_alive(pid):
+            with suppress(ProcessLookupError):
+                os.kill(pid, signal.SIGTERM)
+            deadline = time.time() + max(0.0, timeout_seconds)
+            while time.time() < deadline and _is_alive(pid):
+                time.sleep(0.05)
+            if _is_alive(pid):
+                with suppress(ProcessLookupError):
+                    os.kill(pid, signal.SIGKILL)
+                    signal_name = 'SIGKILL'
+        
+        data['stopped_at'] = _utc_now()
+        data['stop_signal'] = signal_name
+        data['alive'] = False
+        if not self.readonly:
+            _persist_record_state(self._record_path(background_id), data)
+        return data
+
+
 
 def _persist_record_state(path: Path, data: dict[str, Any]) -> None:
     persisted = {k: v for k, v in data.items() if k != 'alive'}
@@ -141,7 +186,12 @@ def _refresh_process_state(data: dict[str, Any], record_path: Path, *, persist:
         return data
 
     pid = int(data['pid'])
-    alive = _is_alive(pid)
+    try:
+        os.kill(pid, 0)
+        alive = True
+    except OSError:
+        alive = False
+
     exit_code: Optional[int] = None
     if alive:
         try:
@@ -165,7 +215,21 @@ def _refresh_process_state(data: dict[str, Any], record_path: Path, *, persist:
     return data
 
 
+def _reap(pid: int) -> bool:
+    try:
+        finished_pid, _ = os.waitpid(pid, os.WNOHANG)
+    except ChildProcessError:
+        return False
+    except OSError:
+        return False
+    return finished_pid == pid
+
+
 def _is_alive(pid: int) -> bool:
+    if pid <= 0:
+        return False
+    if _reap(pid):
+        return False
     try:
         os.kill(pid, 0)
     except OSError: