diff --git a/examples/lark-kanban-control-plane-smoke.py b/examples/lark-kanban-control-plane-smoke.py index 98493e924d..e97c425693 100644 --- a/examples/lark-kanban-control-plane-smoke.py +++ b/examples/lark-kanban-control-plane-smoke.py @@ -15,6 +15,7 @@ sys.path.insert(0, str(REPO_ROOT)) from loopx.extensions.lark.presentation import issue_fix_surface # noqa: E402 +from loopx.extensions.lark.cli_transport import run_command # noqa: E402 from loopx.extensions.lark.presentation.kanban import ( # noqa: E402 CLAIM_UNCLAIMED, STATUS_CLAIMED, @@ -23,7 +24,6 @@ STATUS_REVIEW, STATUS_TODO, LarkKanbanConfig, - _run_command, _lark_record_from_todo_block, build_create_board_plan, default_lark_kanban_config_path, @@ -66,7 +66,7 @@ def forbidden_runner( for argument in ("--ak", "--sk", "--app-secret", "--access-token"): try: - _run_command( + run_command( ["lark-cli", "base", "+record-list", argument, "fixture-secret"], execute=True, runner=forbidden_runner, @@ -77,7 +77,7 @@ def forbidden_runner( raise AssertionError(f"credential argument was accepted: {argument}") assert runner_called is False - dry_run = _run_command( + dry_run = run_command( [ "lark-cli", "base", diff --git a/loopx/chat_lark_api.py b/loopx/chat_lark_api.py index d6f341c7b6..9b3291380c 100644 --- a/loopx/chat_lark_api.py +++ b/loopx/chat_lark_api.py @@ -27,10 +27,7 @@ rebind_lark_manager_session, ) from .extensions.lark.goal_topic_batch import connect_lark_goal_topics -from .extensions.lark.presentation.kanban import ( - CommandRunner, - default_subprocess_runner, -) +from .extensions.lark.cli_transport import CommandRunner, default_subprocess_runner from .chat_agent import CodexChatAgentError from .chat_manager import ( controller_runtime_root, diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 69e11187b8..42ea426056 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -88,7 +88,7 @@ from .extensions.lark.goal_topic_connections import list_lark_apps from .extensions.lark.goal_topic_runtime import LarkGoalTopicRuntimeService from .extensions.lark.manager_routing import authorized_manager_goal_ids -from .extensions.lark.presentation.kanban import CommandRunner +from .extensions.lark.cli_transport import CommandRunner from .extensions.runtime import ( default_extension_state_file, resolve_extension_activation, diff --git a/loopx/cli_commands/explore_feishu_commands.py b/loopx/cli_commands/explore_feishu_commands.py index 370615d1d4..2e3e632e28 100644 --- a/loopx/cli_commands/explore_feishu_commands.py +++ b/loopx/cli_commands/explore_feishu_commands.py @@ -25,7 +25,7 @@ explore_feishu_sync_busy_packet, explore_feishu_sync_singleflight, ) -from ..extensions.lark.presentation.kanban import DEFAULT_CLI_BIN +from ..extensions.lark.cli_transport import DEFAULT_CLI_BIN from ..extensions.lark.presentation.sink_visibility import SINK_VISIBILITY_OWNER_ONLY, SINK_VISIBILITIES diff --git a/loopx/cli_commands/lark_kanban.py b/loopx/cli_commands/lark_kanban.py index 49f7ad7eb0..b622c1f0b7 100644 --- a/loopx/cli_commands/lark_kanban.py +++ b/loopx/cli_commands/lark_kanban.py @@ -7,9 +7,9 @@ from typing import Callable from ..extensions.lark import LARK_EXTENSION_ID, LARK_PROJECTION_SINK_PERMISSION +from ..extensions.lark.cli_transport import DEFAULT_CLI_BIN from ..extensions.lark.presentation.kanban import ( DEFAULT_AGENT_ID, - DEFAULT_CLI_BIN, DEFAULT_STATUS_QUEUE_VIEW, DEFAULT_TABLE_NAME, LarkKanbanConfig, diff --git a/loopx/extensions/lark/card_callback.py b/loopx/extensions/lark/card_callback.py index c58f3f6818..e0c408115a 100644 --- a/loopx/extensions/lark/card_callback.py +++ b/loopx/extensions/lark/card_callback.py @@ -17,7 +17,7 @@ normalized_card_text, ) from .goal_channel_transport import call, json_payload, lark_args -from .presentation.kanban import CommandRunner +from .cli_transport import CommandRunner def callback_timestamp(value: object, *, subject: str) -> str: diff --git a/loopx/extensions/lark/cli_resolution.py b/loopx/extensions/lark/cli_resolution.py index cc0f68c631..1fdac719e6 100644 --- a/loopx/extensions/lark/cli_resolution.py +++ b/loopx/extensions/lark/cli_resolution.py @@ -16,6 +16,8 @@ from pathlib import Path from typing import Any +from .cli_transport import CommandRunner + _SEMVER_DIRECTORY = re.compile(r"^v?(\d+)\.(\d+)\.(\d+)(?:[-+].*)?$") _VERSION_TEXT = re.compile(r"(?:lark-cli\s+)?v?(\d+(?:\.\d+){1,3}(?:[-+][A-Za-z0-9._-]+)?)") @@ -250,7 +252,7 @@ def build_lark_command_runner( *, environ: Mapping[str, str] | None = None, subprocess_run: Callable[..., Any] = subprocess.run, -) -> Callable[[list[str], Path | None, float | None], dict[str, Any]]: +) -> CommandRunner: """Build the dashboard runner with the resolved Node wrapper environment.""" def run( diff --git a/loopx/extensions/lark/cli_transport.py b/loopx/extensions/lark/cli_transport.py new file mode 100644 index 0000000000..d5da7602e6 --- /dev/null +++ b/loopx/extensions/lark/cli_transport.py @@ -0,0 +1,165 @@ +"""Lark provider CLI execution and result decoding. + +Presentation and conversation callers share this IO boundary. Callers still +own authority, retries, and provider readback; a successful process is not a +verified external effect. Executable discovery and Node wrapper environments +remain in cli_resolution. +""" + +from __future__ import annotations + +import json +import shlex +import subprocess +from collections.abc import Callable +from pathlib import Path +from typing import Any + +DEFAULT_CLI_BIN = "lark-cli" +OUTPUT_LIMIT = 1800 +CommandRunner = Callable[[list[str], Path | None, float | None], dict[str, Any]] + +_LARK_CREDENTIAL_ARGUMENTS = frozenset( + { + "--access-key", + "--access-token", + "--ak", + "--app-secret", + "--authorization", + "--bearer-token", + "--client-secret", + "--secret-key", + "--sk", + "--tenant-access-token", + "--user-access-token", + } +) + + +def default_subprocess_runner( + args: list[str], + cwd: Path | None = None, + timeout: float | None = None, +) -> dict[str, Any]: + completed = subprocess.run( + args, + cwd=str(cwd) if cwd else None, + timeout=timeout, + capture_output=True, + text=True, encoding="utf-8", errors="replace", + ) + return { + "returncode": completed.returncode, + "stdout": completed.stdout, + "stderr": completed.stderr, + "timed_out": False, + } + + +def run_command( + args: list[str], + *, + execute: bool, + runner: CommandRunner = default_subprocess_runner, + cwd: Path | None = None, + timeout_seconds: float | None = None, +) -> dict[str, Any]: + credential_argument = next( + ( + argument.split("=", 1)[0].lower() + for argument in args + if argument.split("=", 1)[0].lower() in _LARK_CREDENTIAL_ARGUMENTS + ), + None, + ) + if credential_argument: + raise ValueError( + f"{credential_argument} must remain in lark-cli auth; " + "the Kanban adapter never accepts inline credentials" + ) + command = shlex.join(args) + if not execute: + return { + "command": command, + "executed": False, + "ok": True, + "returncode": None, + "stdout": "", + "stderr": "", + "json": None, + } + try: + result = runner(args, cwd, timeout_seconds) + except subprocess.TimeoutExpired as exc: + return { + "command": command, + "executed": True, + "ok": False, + "returncode": None, + "stdout": str(exc.stdout or ""), + "stderr": str(exc.stderr or ""), + "timed_out": True, + "json": None, + } + except OSError as exc: + return { + "command": command, + "executed": True, + "ok": False, + "returncode": None, + "stdout": "", + "stderr": str(exc), + "json": None, + } + stdout = str(result.get("stdout") or "") + stderr = str(result.get("stderr") or "") + parsed = _parse_json(stdout) + ok = int(result.get("returncode") or 0) == 0 and _parsed_ok(parsed) + return { + "command": command, + "executed": True, + "ok": ok, + "returncode": result.get("returncode"), + "stdout": stdout, + "stderr": stderr, + "json": parsed, + } + + +def _parse_json(text: str) -> Any: + stripped = text.strip() + if not stripped: + return None + try: + return json.loads(stripped) + except json.JSONDecodeError: + return None + + +def _parsed_ok(parsed: Any) -> bool: + if parsed is None: + return True + if not isinstance(parsed, dict): + return True + if parsed.get("ok") is False: + return False + if parsed.get("code") not in (None, 0): + return False + return True + + +def command_error(command_result: dict[str, Any]) -> str: + parsed = command_result.get("json") + if isinstance(parsed, dict): + error = parsed.get("error") + if isinstance(error, dict): + return str(error.get("message") or error) + if error: + return str(error) + if parsed.get("msg") and parsed.get("code") not in (None, 0): + return str(parsed.get("msg")) + stderr = " ".join(str(command_result.get("stderr") or "").split()) + if stderr: + return stderr[:OUTPUT_LIMIT] + stdout = " ".join(str(command_result.get("stdout") or "").split()) + return stdout[:OUTPUT_LIMIT] diff --git a/loopx/extensions/lark/conversation_identity.py b/loopx/extensions/lark/conversation_identity.py index 51cc35fc0e..1bd192964b 100644 --- a/loopx/extensions/lark/conversation_identity.py +++ b/loopx/extensions/lark/conversation_identity.py @@ -14,7 +14,7 @@ from .goal_channel_transport import APP_ID_PATTERN, call, json_payload, lark_args from .identity_shapes import LARK_OPEN_ID_PATTERN as OPEN_ID_PATTERN from .goal_topic_connections import _profile_ref -from .presentation.kanban import CommandRunner +from .cli_transport import CommandRunner def identity_ref(*values: str) -> str: diff --git a/loopx/extensions/lark/goal_channel_blocked_notice.py b/loopx/extensions/lark/goal_channel_blocked_notice.py index 68f1fd5ea7..369e534f62 100644 --- a/loopx/extensions/lark/goal_channel_blocked_notice.py +++ b/loopx/extensions/lark/goal_channel_blocked_notice.py @@ -44,7 +44,7 @@ message_readback_verified, verified_app_id, ) -from .presentation.kanban import ( +from .cli_transport import ( DEFAULT_CLI_BIN, CommandRunner, default_subprocess_runner, diff --git a/loopx/extensions/lark/goal_channel_lifecycle.py b/loopx/extensions/lark/goal_channel_lifecycle.py index 844781a86a..8df206e3b9 100644 --- a/loopx/extensions/lark/goal_channel_lifecycle.py +++ b/loopx/extensions/lark/goal_channel_lifecycle.py @@ -39,10 +39,7 @@ goal_channel_target_for_name, read_goal_channel_targets, ) -from .presentation.kanban import ( - CommandRunner, - default_subprocess_runner, -) +from .cli_transport import CommandRunner, default_subprocess_runner FailureStage = Literal[ "lifecycle", diff --git a/loopx/extensions/lark/goal_channel_message_delivery.py b/loopx/extensions/lark/goal_channel_message_delivery.py index 6bec137da9..c4c54a4b4f 100644 --- a/loopx/extensions/lark/goal_channel_message_delivery.py +++ b/loopx/extensions/lark/goal_channel_message_delivery.py @@ -26,7 +26,7 @@ lark_args, verified_app_id, ) -from .presentation.kanban import CommandRunner +from .cli_transport import CommandRunner class GoalChannelDeliveryStageError(ValueError): diff --git a/loopx/extensions/lark/goal_channel_operation.py b/loopx/extensions/lark/goal_channel_operation.py index c32eee39e4..8cf100f736 100644 --- a/loopx/extensions/lark/goal_channel_operation.py +++ b/loopx/extensions/lark/goal_channel_operation.py @@ -36,7 +36,7 @@ resolve_bound_goal_channel, ) from .identity_shapes import require_card_callback_identity -from .presentation.kanban import CommandRunner, default_subprocess_runner +from .cli_transport import CommandRunner, default_subprocess_runner OPERATION_CARD_ACTION_SCHEMA_VERSION = "loopx_operation_card_action_v0" diff --git a/loopx/extensions/lark/goal_channel_runtime.py b/loopx/extensions/lark/goal_channel_runtime.py index 0dc2209af9..548948b58b 100644 --- a/loopx/extensions/lark/goal_channel_runtime.py +++ b/loopx/extensions/lark/goal_channel_runtime.py @@ -51,13 +51,15 @@ pin_readback_verified, verified_app_id, ) -from .presentation.kanban import ( +from .cli_transport import ( DEFAULT_CLI_BIN, CommandRunner, + default_subprocess_runner, +) +from .presentation.kanban import ( LarkKanbanConfig, build_record_list_command, default_lark_kanban_config_path, - default_subprocess_runner, lark_kanban_config_from_payload, lark_kanban_doctor, read_lark_kanban_local_config, diff --git a/loopx/extensions/lark/goal_channel_setup.py b/loopx/extensions/lark/goal_channel_setup.py index 1a9de22cae..0bf775d86c 100644 --- a/loopx/extensions/lark/goal_channel_setup.py +++ b/loopx/extensions/lark/goal_channel_setup.py @@ -42,11 +42,13 @@ pin_readback_verified, verified_app_id, ) -from .presentation.kanban import ( +from .cli_transport import ( DEFAULT_CLI_BIN, CommandRunner, - default_lark_kanban_config_path, default_subprocess_runner, +) +from .presentation.kanban import ( + default_lark_kanban_config_path, read_lark_kanban_local_config, save_lark_kanban_board_config, setup_lark_kanban_board, diff --git a/loopx/extensions/lark/goal_channel_targets.py b/loopx/extensions/lark/goal_channel_targets.py index c053176330..23de06b883 100644 --- a/loopx/extensions/lark/goal_channel_targets.py +++ b/loopx/extensions/lark/goal_channel_targets.py @@ -11,7 +11,7 @@ from ...file_lock import exclusive_file_lock from .goal_channel_contracts import operation_packet from .goal_channel_transport import APP_ID_PATTERN, CHAT_ID_PATTERN, OPEN_ID_PATTERN -from .presentation.kanban import DEFAULT_CLI_BIN +from .cli_transport import DEFAULT_CLI_BIN from .private_json import write_private_json_atomic GOAL_CHANNEL_TARGETS_SCHEMA_VERSION = "loopx_goal_channel_provider_targets_v0" diff --git a/loopx/extensions/lark/goal_channel_transport.py b/loopx/extensions/lark/goal_channel_transport.py index 39ea408d94..0cdd51e5db 100644 --- a/loopx/extensions/lark/goal_channel_transport.py +++ b/loopx/extensions/lark/goal_channel_transport.py @@ -15,7 +15,7 @@ LARK_MESSAGE_ID_SEARCH as MESSAGE_ID_PATTERN, LARK_OPEN_ID_SEARCH as OPEN_ID_PATTERN, ) -from .presentation.kanban import CommandRunner +from .cli_transport import CommandRunner SAFE_PROFILE_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,99}") REQUIRED_GOAL_TOPIC_SCOPES = ("im:message", "im:message:readonly") diff --git a/loopx/extensions/lark/goal_topic_batch.py b/loopx/extensions/lark/goal_topic_batch.py index 27fcd7fd21..c1e3f7d0c7 100644 --- a/loopx/extensions/lark/goal_topic_batch.py +++ b/loopx/extensions/lark/goal_topic_batch.py @@ -9,7 +9,7 @@ from ...control_plane.todos.contract import normalize_todo_claimed_by from .goal_channel_contracts import operation_packet from .goal_topic_connections import connect_lark_goal_topic -from .presentation.kanban import ( +from .cli_transport import ( DEFAULT_CLI_BIN, CommandRunner, default_subprocess_runner, diff --git a/loopx/extensions/lark/goal_topic_connections.py b/loopx/extensions/lark/goal_topic_connections.py index a88b77d400..bef71637e2 100644 --- a/loopx/extensions/lark/goal_topic_connections.py +++ b/loopx/extensions/lark/goal_topic_connections.py @@ -95,7 +95,7 @@ _connection_routing_modes, decide_lark_topic_route_event, ) -from .presentation.kanban import ( +from .cli_transport import ( DEFAULT_CLI_BIN, CommandRunner, default_subprocess_runner, diff --git a/loopx/extensions/lark/periodic_report_delivery.py b/loopx/extensions/lark/periodic_report_delivery.py index b7597135eb..7025b9f7e4 100644 --- a/loopx/extensions/lark/periodic_report_delivery.py +++ b/loopx/extensions/lark/periodic_report_delivery.py @@ -40,7 +40,7 @@ goal_channel_target_for_name, read_goal_channel_targets, ) -from .presentation.kanban import CommandRunner, default_subprocess_runner +from .cli_transport import CommandRunner, default_subprocess_runner from .presentation.periodic_report import periodic_report_lark_sink_adapter from ...presentation.public_safety import redact_public_text diff --git a/loopx/extensions/lark/presentation/explore_results.py b/loopx/extensions/lark/presentation/explore_results.py index 1b6ce288ff..47443f9df9 100644 --- a/loopx/extensions/lark/presentation/explore_results.py +++ b/loopx/extensions/lark/presentation/explore_results.py @@ -20,6 +20,13 @@ from pathlib import Path from typing import Any, Mapping +from ..cli_transport import ( + DEFAULT_CLI_BIN, + CommandRunner, + command_error, + run_command, + default_subprocess_runner, +) from ....capabilities.explore.result_log import ( EXPLORE_RESULT_PROJECTION_VERSION, FINDING_STATUS_CONFIRMED, @@ -43,16 +50,11 @@ ) from ..private_json import write_private_json_atomic from .kanban import ( - DEFAULT_CLI_BIN, - CommandRunner, - _command_error, _extract_base_token, _extract_table_id, _extract_created_record_id, _public_safe_text, - _run_command, _select_options, - default_subprocess_runner, lark_record_rows, now_lark_datetime, parse_lark_base_url, @@ -629,7 +631,7 @@ def _auto_failure(error: str) -> dict[str, Any]: {"block_token": str(stored_canonical["overview_whiteboard_token"])}, ] elif not document_token: - create_command = _run_command( + create_command = run_command( [ cli_bin, "docs", @@ -645,7 +647,7 @@ def _auto_failure(error: str) -> dict[str, Any]: runner=default_subprocess_runner, ) if not create_command.get("ok"): - return _auto_failure(_command_error(create_command)) + return _auto_failure(command_error(create_command)) document_token = str( ( (create_command.get("json") or {}) @@ -661,7 +663,7 @@ def _auto_failure(error: str) -> dict[str, Any]: ) overview_boards = _created_overview_boards(create_command) else: - append_command = _run_command( + append_command = run_command( [ cli_bin, "docs", @@ -681,7 +683,7 @@ def _auto_failure(error: str) -> dict[str, Any]: runner=default_subprocess_runner, ) if not append_command.get("ok"): - return _auto_failure(_command_error(append_command)) + return _auto_failure(command_error(append_command)) overview_boards = _created_overview_boards(append_command) if len(overview_boards) < 2: return _auto_failure( @@ -1017,7 +1019,7 @@ def sync_explore_visual_to_lark( if delivery_ok else str(readback.get("error") or "") if result.get("ok") - else _command_error(result) + else command_error(result) ), } @@ -1373,13 +1375,13 @@ def failure(error: str | None = None) -> dict[str, Any]: "warnings": warnings, "error": error or next( - (_command_error(item) for item in commands if not item.get("ok")), + (command_error(item) for item in commands if not item.get("ok")), "unknown", ), } if not effective_base_token: - create = _run_command( + create = run_command( [cli_bin, "base", "+base-create", "--as", identity, "--name", base_name], execute=execute, runner=runner, @@ -1404,7 +1406,7 @@ def failure(error: str | None = None) -> dict[str, Any]: for table_key in EXPLORE_TABLE_KEYS: if table_ids.get(table_key): continue - table_create = _run_command( + table_create = run_command( [ cli_bin, "base", @@ -1637,7 +1639,7 @@ def sync_explore_results_to_lark( for scan_goal_id in scan_goal_ids: offset = 0 while True: - list_result = _run_command( + list_result = run_command( _build_record_list_command( config, table_id=config.table_id(table_key), @@ -1702,7 +1704,7 @@ def sync_explore_results_to_lark( result = _skipped_sync_command() skipped_rows += 1 else: - result = _run_command( + result = run_command( _build_upsert_command( config, table_id=config.table_id(table_key), @@ -1762,7 +1764,7 @@ def sync_explore_results_to_lark( continue offset = 0 while True: - list_result = _run_command( + list_result = run_command( _build_record_list_command( config, table_id=config.table_id(table_key), @@ -1868,7 +1870,7 @@ def sync_explore_results_to_lark( "row/result-id readback failed" if readback.get("performed") and not readback_verified else next( - (_command_error(item) for item in commands if not item.get("ok")), + (command_error(item) for item in commands if not item.get("ok")), "unknown", ) ), diff --git a/loopx/extensions/lark/presentation/explore_stage_document.py b/loopx/extensions/lark/presentation/explore_stage_document.py index 54d3b05f2b..69570e5dbc 100644 --- a/loopx/extensions/lark/presentation/explore_stage_document.py +++ b/loopx/extensions/lark/presentation/explore_stage_document.py @@ -10,7 +10,11 @@ from pathlib import Path from typing import Any, Protocol -from .kanban import CommandRunner, _command_error, _run_command +from ..cli_transport import ( + CommandRunner, + command_error, + run_command, +) class StageDocumentConfig(Protocol): @@ -279,12 +283,12 @@ def observe(command: Mapping[str, Any], *, action: str) -> None: "action": action, "ok": bool(command.get("ok")), "returncode": command.get("returncode"), - "error": None if command.get("ok") else _command_error(command), + "error": None if command.get("ok") else command_error(command), } ) if execute and docx_token: - outline_command = _run_command( + outline_command = run_command( _docs_fetch_command( config, docx_token=docx_token, @@ -307,7 +311,7 @@ def observe(command: Mapping[str, Any], *, action: str) -> None: [], reconciliation_commands, reconciliation, - _command_error(outline_command) + command_error(outline_command) if not outline_command.get("ok") else "Evidence Stage outline could not be parsed", ) @@ -368,7 +372,7 @@ def observe(command: Mapping[str, Any], *, action: str) -> None: ) for stage_index in indexes_to_inspect: for heading_block_id in headings_by_stage.get(stage_index, []): - section_command = _run_command( + section_command = run_command( _docs_fetch_command( config, docx_token=docx_token, @@ -399,7 +403,7 @@ def observe(command: Mapping[str, Any], *, action: str) -> None: [], reconciliation_commands, reconciliation, - _command_error(section_command) + command_error(section_command) if not section_command.get("ok") else ( f"Evidence Stage {stage_index:02d} conflicts with an " @@ -475,7 +479,7 @@ def observe(command: Mapping[str, Any], *, action: str) -> None: if revision is not None: delete_args.extend(["--revision-id", str(revision)]) delete_args.extend(["--format", "json"]) - delete_command = _run_command( + delete_command = run_command( delete_args, execute=True, runner=runner, @@ -489,9 +493,9 @@ def observe(command: Mapping[str, Any], *, action: str) -> None: [], reconciliation_commands, reconciliation, - _command_error(delete_command), + command_error(delete_command), ) - verification_command = _run_command( + verification_command = run_command( _docs_fetch_command( config, docx_token=docx_token, @@ -514,7 +518,7 @@ def observe(command: Mapping[str, Any], *, action: str) -> None: [], reconciliation_commands, reconciliation, - _command_error(verification_command) + command_error(verification_command) if not verification_command.get("ok") else "updated Evidence Stage outline could not be parsed", ) @@ -582,7 +586,7 @@ def observe(command: Mapping[str, Any], *, action: str) -> None: section_commands: list[dict[str, Any]] = [] for stage in missing: stage_index = int(stage.get("stage_index") or 0) - command = _run_command( + command = run_command( [ config.cli_bin, "docs", @@ -609,7 +613,7 @@ def observe(command: Mapping[str, Any], *, action: str) -> None: section_commands, reconciliation_commands, reconciliation, - _command_error(command), + command_error(command), ) if execute: section = _created_stage_section(command) @@ -640,7 +644,7 @@ def observe(command: Mapping[str, Any], *, action: str) -> None: } if execute and docx_token and missing: - verification_command = _run_command( + verification_command = run_command( _docs_fetch_command( config, docx_token=docx_token, @@ -663,7 +667,7 @@ def observe(command: Mapping[str, Any], *, action: str) -> None: section_commands, reconciliation_commands, reconciliation, - _command_error(verification_command) + command_error(verification_command) if not verification_command.get("ok") else "created Evidence Stage outline could not be parsed", ) diff --git a/loopx/extensions/lark/presentation/explore_visual_readback.py b/loopx/extensions/lark/presentation/explore_visual_readback.py index 2a4192d357..3c35745873 100644 --- a/loopx/extensions/lark/presentation/explore_visual_readback.py +++ b/loopx/extensions/lark/presentation/explore_visual_readback.py @@ -8,7 +8,11 @@ from pathlib import Path from typing import Any, Mapping, Sequence -from .kanban import CommandRunner, _command_error, _run_command +from ..cli_transport import ( + CommandRunner, + command_error, + run_command, +) VISUAL_READBACK_RETRY_DELAYS_SECONDS = (0.25, 0.5, 1.0, 2.0, 4.0) @@ -86,7 +90,7 @@ def readback_visual_delivery_marker( else retry_delays ) for attempt_index in range(len(effective_retry_delays) + 1): - result = _run_command( + result = run_command( command, execute=True, runner=runner, @@ -152,7 +156,7 @@ def readback_visual_delivery_marker( "error": ( None if result.get("ok") and marker_observed - else _command_error(result) + else command_error(result) if not result.get("ok") else "remote whiteboard raw nodes do not contain the expected delivery marker" ), @@ -195,16 +199,16 @@ def publish_visual_with_readback( not reconciled_existing_delivery and not prepublish_command.get("ok") ) if not execute: - result = _run_command(command, execute=False, runner=runner) + result = run_command(command, execute=False, runner=runner) publish_attempts = [result] elif reconciled_existing_delivery or prepublish_readback_blocked: - result = _run_command(command, execute=False, runner=runner) + result = run_command(command, execute=False, runner=runner) publish_attempts = [] else: source_path.parent.mkdir(parents=True, exist_ok=True) source_path.write_text(published_source, encoding="utf-8") try: - result = _run_command( + result = run_command( command, execute=True, runner=runner, @@ -225,7 +229,7 @@ def publish_visual_with_readback( retry_command[token_index] = ( "loopx-retry-" + hashlib.sha256(retry_material).hexdigest()[:32] ) - result = _run_command( + result = run_command( retry_command, execute=True, runner=runner, diff --git a/loopx/extensions/lark/presentation/kanban.py b/loopx/extensions/lark/presentation/kanban.py index d6c28dd850..b359bcb6aa 100644 --- a/loopx/extensions/lark/presentation/kanban.py +++ b/loopx/extensions/lark/presentation/kanban.py @@ -3,11 +3,10 @@ import json import re import shlex -import subprocess from dataclasses import dataclass from datetime import datetime from pathlib import Path -from typing import Any, Callable, Optional +from typing import Any from urllib.parse import parse_qs, urlparse from ....control_plane.todos.contract import ( @@ -25,6 +24,7 @@ plan_projection_source_reconcile, validate_projection_source_reconcile_request, ) +from .. import cli_transport from ..private_json import write_private_json_atomic from . import issue_fix_surface from .projection_rows import ( @@ -55,24 +55,8 @@ LARK_KANBAN_SYNC_PROJECTION_VERSION = "loopx_lark_kanban_sync_projection_v0" DEFAULT_TABLE_NAME = "LoopX Control Plane" DEFAULT_AGENT_ID = "codex-kanban-worker" -DEFAULT_CLI_BIN = "lark-cli" DEFAULT_STATUS_QUEUE_VIEW = "Worker Queue" OPERATOR_CARD_FIELDS = ["Task", "Claim", "Priority", "User Gate", "Evidence", "Status"] -_LARK_CREDENTIAL_ARGUMENTS = frozenset( - { - "--access-key", - "--access-token", - "--ak", - "--app-secret", - "--authorization", - "--bearer-token", - "--client-secret", - "--secret-key", - "--sk", - "--tenant-access-token", - "--user-access-token", - } -) STATUS_TODO = "Todo" STATUS_CLAIMED = "Claimed" @@ -86,10 +70,6 @@ CLAIM_AGENT = "Agent" TEXT_LIMIT = 4000 -OUTPUT_LIMIT = 1800 - - -CommandRunner = Callable[[list[str], Optional[Path], Optional[float]], dict[str, Any]] def _select_options(names: list[str]) -> list[dict[str, str]]: @@ -453,135 +433,6 @@ def lark_kanban_feasibility_cases( ] -def default_subprocess_runner( - args: list[str], - cwd: Path | None = None, - timeout: float | None = None, -) -> dict[str, Any]: - completed = subprocess.run( - args, - cwd=str(cwd) if cwd else None, - timeout=timeout, - capture_output=True, - text=True, encoding="utf-8", errors="replace", - ) - return { - "returncode": completed.returncode, - "stdout": completed.stdout, - "stderr": completed.stderr, - "timed_out": False, - } - - -def _run_command( - args: list[str], - *, - execute: bool, - runner: CommandRunner = default_subprocess_runner, - cwd: Path | None = None, - timeout_seconds: float | None = None, -) -> dict[str, Any]: - credential_argument = next( - ( - argument.split("=", 1)[0].lower() - for argument in args - if argument.split("=", 1)[0].lower() in _LARK_CREDENTIAL_ARGUMENTS - ), - None, - ) - if credential_argument: - raise ValueError( - f"{credential_argument} must remain in lark-cli auth; " - "the Kanban adapter never accepts inline credentials" - ) - command = shlex.join(args) - if not execute: - return { - "command": command, - "executed": False, - "ok": True, - "returncode": None, - "stdout": "", - "stderr": "", - "json": None, - } - try: - result = runner(args, cwd, timeout_seconds) - except subprocess.TimeoutExpired as exc: - return { - "command": command, - "executed": True, - "ok": False, - "returncode": None, - "stdout": str(exc.stdout or ""), - "stderr": str(exc.stderr or ""), - "timed_out": True, - "json": None, - } - except OSError as exc: - return { - "command": command, - "executed": True, - "ok": False, - "returncode": None, - "stdout": "", - "stderr": str(exc), - "json": None, - } - stdout = str(result.get("stdout") or "") - stderr = str(result.get("stderr") or "") - parsed = _parse_json(stdout) - ok = int(result.get("returncode") or 0) == 0 and _parsed_ok(parsed) - return { - "command": command, - "executed": True, - "ok": ok, - "returncode": result.get("returncode"), - "stdout": stdout, - "stderr": stderr, - "json": parsed, - } - - -def _parse_json(text: str) -> Any: - stripped = text.strip() - if not stripped: - return None - try: - return json.loads(stripped) - except json.JSONDecodeError: - return None - - -def _parsed_ok(parsed: Any) -> bool: - if parsed is None: - return True - if not isinstance(parsed, dict): - return True - if parsed.get("ok") is False: - return False - if parsed.get("code") not in (None, 0): - return False - return True - - -def _command_error(command_result: dict[str, Any]) -> str: - parsed = command_result.get("json") - if isinstance(parsed, dict): - error = parsed.get("error") - if isinstance(error, dict): - return str(error.get("message") or error) - if error: - return str(error) - if parsed.get("msg") and parsed.get("code") not in (None, 0): - return str(parsed.get("msg")) - stderr = " ".join(str(command_result.get("stderr") or "").split()) - if stderr: - return stderr[:OUTPUT_LIMIT] - stdout = " ".join(str(command_result.get("stdout") or "").split()) - return stdout[:OUTPUT_LIMIT] - - def _compact_text(value: Any, *, limit: int = TEXT_LIMIT) -> str: text = " ".join(str(value or "").split()) if len(text) <= limit: @@ -682,7 +533,7 @@ class LarkKanbanConfig: base_token: str table_id: str view_id: str | None = DEFAULT_STATUS_QUEUE_VIEW - cli_bin: str = DEFAULT_CLI_BIN + cli_bin: str = cli_transport.DEFAULT_CLI_BIN identity: str = "user" @@ -777,7 +628,7 @@ def save_lark_kanban_board_config( base_url: str | None = None, base_name: str | None = None, table_name: str | None = None, - cli_bin: str = DEFAULT_CLI_BIN, + cli_bin: str = cli_transport.DEFAULT_CLI_BIN, identity: str = "user", view_ids: dict[str, str] | None = None, merge_existing: bool = True, @@ -833,7 +684,7 @@ def lark_kanban_config_from_payload(payload: dict[str, Any]) -> LarkKanbanConfig **{"base_" + "token": base_token}, table_id=table_id, view_id=str(board.get("view_id") or DEFAULT_STATUS_QUEUE_VIEW), - cli_bin=str(board.get("cli_bin") or DEFAULT_CLI_BIN), + cli_bin=str(board.get("cli_bin") or cli_transport.DEFAULT_CLI_BIN), identity=str(board.get("identity") or "user"), ) @@ -899,7 +750,7 @@ def build_create_board_plan( *, base_name: str, table_name: str, - cli_bin: str = DEFAULT_CLI_BIN, + cli_bin: str = cli_transport.DEFAULT_CLI_BIN, identity: str = "bot", base_token: str | None = None, user_open_id: str | None = None, @@ -1022,18 +873,18 @@ def create_lark_kanban_board( *, base_name: str, table_name: str = DEFAULT_TABLE_NAME, - cli_bin: str = DEFAULT_CLI_BIN, + cli_bin: str = cli_transport.DEFAULT_CLI_BIN, identity: str = "bot", base_token: str | None = None, user_open_id: str | None = None, execute: bool = False, - runner: CommandRunner = default_subprocess_runner, + runner: cli_transport.CommandRunner = cli_transport.default_subprocess_runner, ) -> dict[str, Any]: commands: list[dict[str, Any]] = [] effective_base_token = base_token table_id: str | None = None if not effective_base_token: - create = _run_command( + create = cli_transport.run_command( [cli_bin, "base", "+base-create", "--as", identity, "--name", base_name], execute=execute, runner=runner, @@ -1057,7 +908,7 @@ def create_lark_kanban_board( ) return _board_payload(False, commands, effective_base_token, table_id) token = effective_base_token or "" - table_create = _run_command( + table_create = cli_transport.run_command( [ cli_bin, "base", @@ -1083,7 +934,7 @@ def create_lark_kanban_board( table_id = _extract_table_id(table_create.get("json")) or table_name table_ref = table_id or table_name if user_open_id: - grant = _run_command( + grant = cli_transport.run_command( [ cli_bin, "drive", @@ -1119,7 +970,7 @@ def create_lark_kanban_board( if grant.get("executed") and not grant.get("ok"): return _board_payload(False, commands, effective_base_token, table_id) for command in _view_configuration_commands(cli_bin=cli_bin, identity=identity, base_token=token, table_id=table_ref, view_ids={view["name"]: view["name"] for view in lark_kanban_views()}): - result = _run_command(command, execute=execute, runner=runner) + result = cli_transport.run_command(command, execute=execute, runner=runner) commands.append(result) if result.get("executed") and not result.get("ok"): return _board_payload(False, commands, effective_base_token, table_id) @@ -1138,7 +989,7 @@ def _board_payload( "base_token": base_token, "table_id": table_id, "commands": commands, - "error": None if ok else next((_command_error(item) for item in commands if not item.get("ok")), "unknown"), + "error": None if ok else next((cli_transport.command_error(item) for item in commands if not item.get("ok")), "unknown"), } @@ -1267,7 +1118,7 @@ def use_lark_kanban_board( base_token: str | None = None, table_id: str | None = None, view_id: str | None = None, - cli_bin: str = DEFAULT_CLI_BIN, + cli_bin: str = cli_transport.DEFAULT_CLI_BIN, identity: str = "user", ) -> dict[str, Any]: parsed = parse_lark_base_url(base_url) if base_url else {} @@ -1293,10 +1144,10 @@ def setup_lark_kanban_board( base_url: str | None = None, base_token: str | None = None, table_id: str | None = None, - cli_bin: str = DEFAULT_CLI_BIN, + cli_bin: str = cli_transport.DEFAULT_CLI_BIN, identity: str = "user", execute: bool = False, - runner: CommandRunner = default_subprocess_runner, + runner: cli_transport.CommandRunner = cli_transport.default_subprocess_runner, ) -> dict[str, Any]: commands: list[dict[str, Any]] = [] warnings: list[str] = [] @@ -1349,7 +1200,7 @@ def save_usable_config() -> None: ) def partial_enrichment_payload(command_result: dict[str, Any]) -> dict[str, Any]: - error = _command_error(command_result) + error = cli_transport.command_error(command_result) warnings.append(f"usable board config saved; view enrichment skipped: {error}") return _setup_payload( True, @@ -1366,7 +1217,7 @@ def partial_enrichment_payload(command_result: dict[str, Any]) -> dict[str, Any] if not effective_base_token: created_base = True - create = _run_command( + create = cli_transport.run_command( [ cli_bin, "base", @@ -1404,7 +1255,7 @@ def partial_enrichment_payload(command_result: dict[str, Any]) -> dict[str, Any] effective_base_token = "" if not effective_table_id: if created_base: - blocks = _run_command( + blocks = cli_transport.run_command( [ cli_bin, "base", @@ -1439,7 +1290,7 @@ def partial_enrichment_payload(command_result: dict[str, Any]) -> dict[str, Any] effective_table_id = "" else: created_table = True - table_create = _run_command( + table_create = cli_transport.run_command( [ cli_bin, "base", @@ -1480,7 +1331,7 @@ def partial_enrichment_payload(command_result: dict[str, Any]) -> dict[str, Any] save_usable_config() if execute and not created_base and not created_table: - field_list = _run_command( + field_list = cli_transport.run_command( [cli_bin, "base", "+field-list", "--as", identity, "--base-token", effective_base_token, "--table-id", effective_table_id, "--format", "json", "--limit", "200"], execute=True, runner=runner, @@ -1489,7 +1340,7 @@ def partial_enrichment_payload(command_result: dict[str, Any]) -> dict[str, Any] if not field_list.get("ok"): return partial_enrichment_payload(field_list) for definition in issue_fix_surface.missing_field_definitions(field_list.get("json"), lark_kanban_field_definitions()): - field_create = _run_command( + field_create = cli_transport.run_command( [cli_bin, "base", "+field-create", "--as", identity, "--base-token", effective_base_token, "--table-id", effective_table_id, "--json", json.dumps(definition, ensure_ascii=False)], execute=True, runner=runner, @@ -1501,7 +1352,7 @@ def partial_enrichment_payload(command_result: dict[str, Any]) -> dict[str, Any] for migration in issue_fix_surface.field_definition_migrations( field_list.get("json"), lark_kanban_field_definitions() ): - field_update = _run_command( + field_update = cli_transport.run_command( [ cli_bin, "base", @@ -1542,7 +1393,7 @@ def partial_enrichment_payload(command_result: dict[str, Any]) -> dict[str, Any] missing_view_names = [view["name"] for view in lark_kanban_views() if view["name"] not in view_ids] should_create_views = bool(missing_view_names) and (created_base or not created_table) if should_create_views: - view_create = _run_command( + view_create = cli_transport.run_command( [ cli_bin, "base", @@ -1587,7 +1438,7 @@ def partial_enrichment_payload(command_result: dict[str, Any]) -> dict[str, Any] table_id=effective_table_id, view_ids=view_ids, ): - result = _run_command(command, execute=execute, runner=runner) + result = cli_transport.run_command(command, execute=execute, runner=runner) commands.append(result) if execute and not result.get("ok"): if config_payload: @@ -1646,11 +1497,11 @@ def _setup_payload( payload["partial"] = True payload["enrichment_ok"] = False payload["enrichment_error"] = enrichment_error or next( - (_command_error(item) for item in commands if not item.get("ok")), + (cli_transport.command_error(item) for item in commands if not item.get("ok")), "unknown", ) elif not ok: - payload["error"] = error or next((_command_error(item) for item in commands if not item.get("ok")), "unknown") + payload["error"] = error or next((cli_transport.command_error(item) for item in commands if not item.get("ok")), "unknown") return payload @@ -1661,9 +1512,9 @@ def _refresh_view_ids( base_token: str, table_id: str, commands: list[dict[str, Any]], - runner: CommandRunner, + runner: cli_transport.CommandRunner, ) -> dict[str, str]: - view_list = _run_command( + view_list = cli_transport.run_command( [ cli_bin, "base", @@ -1720,9 +1571,9 @@ def _inspect_lark_kanban_schema( config: LarkKanbanConfig, *, commands: list[dict[str, Any]], - runner: CommandRunner, + runner: cli_transport.CommandRunner, ) -> dict[str, Any]: - field_list = _run_command( + field_list = cli_transport.run_command( [ config.cli_bin, "base", @@ -1750,7 +1601,7 @@ def _inspect_lark_kanban_schema( "missing_fields": [], "incompatible_fields": [], "remediation_command": "loopx lark-kanban doctor --require-board", - "error": _command_error(field_list), + "error": cli_transport.command_error(field_list), } return _lark_kanban_schema_check(field_list.get("json")) @@ -1775,11 +1626,11 @@ def _schema_check_error(schema_check: dict[str, Any]) -> str: def lark_kanban_doctor( *, config_path: Path, - cli_bin: str = DEFAULT_CLI_BIN, + cli_bin: str = cli_transport.DEFAULT_CLI_BIN, identity: str = "user", check_board: bool = True, require_board: bool = False, - runner: CommandRunner = default_subprocess_runner, + runner: cli_transport.CommandRunner = cli_transport.default_subprocess_runner, ) -> dict[str, Any]: commands: list[dict[str, Any]] = [] issues: list[dict[str, str]] = [] @@ -1828,10 +1679,10 @@ def lark_kanban_doctor( board_config.table_id, ], ): - result = _run_command(command, execute=True, runner=runner) + result = cli_transport.run_command(command, execute=True, runner=runner) commands.append(result) if not result.get("ok"): - issues.append({"severity": "error", "message": _command_error(result)}) + issues.append({"severity": "error", "message": cli_transport.command_error(result)}) if not any(issue["severity"] == "error" for issue in issues): schema_check = _inspect_lark_kanban_schema( board_config, @@ -1862,17 +1713,17 @@ def _lark_kanban_preflight( *, cli_bin: str, identity: str, - runner: CommandRunner, + runner: cli_transport.CommandRunner, ) -> dict[str, Any]: commands: list[dict[str, Any]] = [] warnings: list[str] = [] - version = _run_command([cli_bin, "--version"], execute=True, runner=runner) + version = cli_transport.run_command([cli_bin, "--version"], execute=True, runner=runner) commands.append(version) cli_ok = bool(version.get("ok")) version_tuple = _version_tuple(str(version.get("stdout") or "")) if cli_ok and version_tuple and version_tuple < (1, 0, 56): warnings.append("lark-cli should be upgraded to at least 1.0.56 for setup visible-field support") - auth = _run_command([cli_bin, "auth", "status"], execute=True, runner=runner) + auth = cli_transport.run_command([cli_bin, "auth", "status"], execute=True, runner=runner) commands.append(auth) identity_available = _identity_available(auth.get("json"), identity) for help_command in ( @@ -1882,7 +1733,7 @@ def _lark_kanban_preflight( [cli_bin, "base", "+view-set-group", "--help"], [cli_bin, "base", "+view-set-visible-fields", "--help"], ): - result = _run_command(help_command, execute=True, runner=runner) + result = cli_transport.run_command(help_command, execute=True, runner=runner) commands.append(result) if not result.get("ok"): warnings.append(f"missing lark-cli shortcut: {shlex.join(help_command[2:-1])}") @@ -1985,7 +1836,7 @@ def sync_loopx_projection_to_lark_kanban( reconcile_source: bool = False, source_snapshot_complete: bool = False, execute: bool = False, - runner: CommandRunner = default_subprocess_runner, + runner: cli_transport.CommandRunner = cli_transport.default_subprocess_runner, ) -> dict[str, Any]: if not isinstance(projection, dict): raise ValueError("projection must be a JSON object") @@ -2102,7 +1953,7 @@ def sync_loopx_projection_to_lark_kanban( source_id=resolved_source_id, public_safe=public_safe, ) - result = _run_command( + result = cli_transport.run_command( build_record_upsert_command(config, record_id=record_map.get(key), values=values), execute=execute, runner=runner, @@ -2125,7 +1976,7 @@ def sync_loopx_projection_to_lark_kanban( executed_remote_delete_count = 0 if ok and reconcile_source and reconcile_plan["remote_delete_record_ids"]: - delete_result = _run_command( + delete_result = cli_transport.run_command( build_record_delete_command( config, record_ids=reconcile_plan["remote_delete_record_ids"], @@ -2189,7 +2040,7 @@ def sync_loopx_todos_to_lark_kanban( include_done: bool = False, limit: int = 50, execute: bool = False, - runner: CommandRunner = default_subprocess_runner, + runner: cli_transport.CommandRunner = cli_transport.default_subprocess_runner, ) -> dict[str, Any]: from ....capabilities.issue_fix.outcome_projection import ( build_issue_fix_outcome_collection_from_domain_state, @@ -2302,7 +2153,7 @@ def sync_loopx_todos_to_lark_kanban( state_file=resolved_state_file, priority=todo_priority_label(block, text_mode="prefix") or "P2", ) - result = _run_command( + result = cli_transport.run_command( build_record_upsert_command(config, record_id=record_map.get(key), values=values), execute=execute, runner=runner, @@ -2338,7 +2189,7 @@ def sync_loopx_todos_to_lark_kanban( goal_id=row_goal_id, source_id=outcome_source_id, ) - result = _run_command( + result = cli_transport.run_command( build_record_upsert_command( config, record_id=record_map.get(key), @@ -2403,7 +2254,7 @@ def sync_loopx_todos_to_lark_kanban( def _load_lark_todo_record_map( config: LarkKanbanConfig, *, config_path: Path | None, execute: bool, inspect_remote: bool, - commands: list[dict[str, Any]], runner: CommandRunner, + commands: list[dict[str, Any]], runner: cli_transport.CommandRunner, ) -> tuple[dict[str, Any], dict[str, str], list[dict[str, str]], bool | None]: local = read_lark_kanban_local_config(config_path) if config_path else {} record_map = _todo_record_map_from_local_config(local) @@ -2417,7 +2268,7 @@ def _load_lark_todo_record_map( cli_bin=config.cli_bin, identity=config.identity, ) - list_result = _run_command(build_record_list_command(list_config), execute=True, runner=runner) + list_result = cli_transport.run_command(build_record_list_command(list_config), execute=True, runner=runner) commands.append(list_result) remote_records: list[dict[str, str]] = [] remote_complete = False @@ -2533,9 +2384,9 @@ def seed_lark_kanban_task( *, task: dict[str, Any], execute: bool = False, - runner: CommandRunner = default_subprocess_runner, + runner: cli_transport.CommandRunner = cli_transport.default_subprocess_runner, ) -> dict[str, Any]: - result = _run_command( + result = cli_transport.run_command( build_record_upsert_command(config, record_id=None, values=task), execute=execute, runner=runner, @@ -2555,7 +2406,7 @@ def seed_lark_kanban_records( *, records: list[dict[str, Any]], execute: bool = False, - runner: CommandRunner = default_subprocess_runner, + runner: cli_transport.CommandRunner = cli_transport.default_subprocess_runner, ) -> dict[str, Any]: results: list[dict[str, Any]] = [] ok = True @@ -2588,13 +2439,13 @@ def lark_kanban_heartbeat( execute_worker: bool = False, complete_on_success: bool = False, allowed_command_prefixes: list[str] | None = None, - runner: CommandRunner = default_subprocess_runner, + runner: cli_transport.CommandRunner = cli_transport.default_subprocess_runner, now: datetime | None = None, worker_timeout_seconds: float = 600.0, ) -> dict[str, Any]: commands: list[dict[str, Any]] = [] if fixture is None: - list_result = _run_command( + list_result = cli_transport.run_command( build_record_list_command(config), execute=execute_lark, runner=runner, @@ -2639,7 +2490,7 @@ def lark_kanban_heartbeat( "Last Heartbeat": timestamp, "Run History": append_history(task.get("Run History"), f"{timestamp} {agent_id}: claimed"), } - claim_result = _run_command( + claim_result = cli_transport.run_command( build_record_upsert_command(config, record_id=record_id, values=claim_values), execute=execute_lark, runner=runner, @@ -2692,7 +2543,7 @@ def lark_kanban_heartbeat( else: workdir = Path(str(task.get("Workdir") or ".")).expanduser() worker_args = shlex.split(command) - worker_result = _run_command( + worker_result = cli_transport.run_command( worker_args, execute=True, runner=runner, @@ -2706,15 +2557,15 @@ def lark_kanban_heartbeat( "reason": "completed" if worker_result.get("ok") else "failed", "command": command, "returncode": worker_result.get("returncode"), - "stdout": _compact_text(worker_result.get("stdout"), limit=OUTPUT_LIMIT), - "stderr": _compact_text(worker_result.get("stderr"), limit=OUTPUT_LIMIT), + "stdout": _compact_text(worker_result.get("stdout"), limit=cli_transport.OUTPUT_LIMIT), + "stderr": _compact_text(worker_result.get("stderr"), limit=cli_transport.OUTPUT_LIMIT), } final_status = STATUS_DONE if worker_result.get("ok") and complete_on_success else STATUS_REVIEW if not worker_result.get("ok"): final_status = STATUS_BLOCKED evidence = _compact_text( worker.get("stdout") or worker.get("stderr") or worker.get("reason"), - limit=OUTPUT_LIMIT, + limit=cli_transport.OUTPUT_LIMIT, ) result_code = worker_result.get("returncode") finish_values = { @@ -2731,7 +2582,7 @@ def lark_kanban_heartbeat( ), "Handoff": _handoff_for_status(final_status, agent_id=agent_id), } - finish_result = _run_command( + finish_result = cli_transport.run_command( build_record_upsert_command(config, record_id=record_id, values=finish_values), execute=execute_lark, runner=runner, diff --git a/loopx/extensions/lark/team_plan_confirmation.py b/loopx/extensions/lark/team_plan_confirmation.py index 8591929660..9de7eb65f7 100644 --- a/loopx/extensions/lark/team_plan_confirmation.py +++ b/loopx/extensions/lark/team_plan_confirmation.py @@ -55,7 +55,7 @@ validate_team_plan_delivery_receipt, write_delivery as write_manager_delivery, ) -from .presentation.kanban import CommandRunner, default_subprocess_runner +from .cli_transport import CommandRunner, default_subprocess_runner from .presentation.team_plan import ( TEAM_PLAN_CARD_ACTION_SCHEMA_VERSION, build_team_plan_result_card, diff --git a/tests/extensions/test_lark_cli_transport.py b/tests/extensions/test_lark_cli_transport.py new file mode 100644 index 0000000000..77f7af12f7 --- /dev/null +++ b/tests/extensions/test_lark_cli_transport.py @@ -0,0 +1,135 @@ +"""Characterize the provider command boundary with real local subprocesses.""" + +import json +from pathlib import Path +import shlex +import shutil +import subprocess +import sys + +import pytest + +from loopx.extensions.lark.cli_transport import ( + command_error, + run_command, + default_subprocess_runner, +) + + +def test_preview_does_not_start_a_process(tmp_path: Path) -> None: + marker = tmp_path / "started" + command = [sys.executable, "-c", "from pathlib import Path; Path('started').touch()"] + result = run_command(command, execute=False, cwd=tmp_path) + assert result["ok"] is True + assert result["executed"] is False + assert result["command"] == shlex.join(command) + assert not marker.exists() + + +@pytest.mark.parametrize("execute", [False, True]) +@pytest.mark.parametrize("argument", ["--access-token", "--APP-SECRET=fixture"]) +def test_inline_credentials_refuse_before_execution(argument: str, execute: bool) -> None: + def forbidden(*_args): + pytest.fail("credential refusal must precede provider execution") + + with pytest.raises(ValueError, match="must remain in lark-cli auth"): + run_command(["lark-cli", argument], execute=execute, runner=forbidden) + + +def test_real_subprocess_preserves_cwd_and_decodes_utf8(tmp_path: Path) -> None: + command = [sys.executable, "-c", ( + "import os,sys; from pathlib import Path; " + "Path('started').touch(); " + "os.write(1, b'{\"ok\":true,\"message\":\"caf\\xc3\\xa9 \\xff\"}'); " + "os.write(2, '诊断'.encode('utf-8'))" + )] + result = run_command(command, execute=True, cwd=tmp_path, timeout_seconds=5) + assert result["ok"] is True + assert result["json"] == {"ok": True, "message": "café �"} + assert result["stderr"] == "诊断" + assert (tmp_path / "started").exists() + raw = default_subprocess_runner([sys.executable, "-c", "print('complete')"]) + assert raw == {"returncode": 0, "stdout": "complete\n", "stderr": "", "timed_out": False} + + +@pytest.mark.parametrize(("body", "exit_code", "expected_ok"), [ + ('{"ok": false}', 0, False), + ('{"code": 42, "msg": "provider rejected"}', 0, False), + ('{"ok": true, "code": 0}', 3, False), + ('{"code": 0}', 0, True), + ('[1, 2]', 0, True), + ('plain provider output', 0, True), +]) +def test_process_and_provider_failures_remain_distinct(body, exit_code, expected_ok): + result = run_command([ + sys.executable, "-c", "import sys; print(sys.argv[1]); sys.exit(int(sys.argv[2]))", + body, str(exit_code), + ], execute=True, timeout_seconds=5) + assert result["ok"] is expected_ok + assert result["returncode"] == exit_code + assert result["stdout"].strip() == body + assert result["json"] == (None if body == "plain provider output" else json.loads(body)) + + +def test_timeout_preserves_started_observation_and_missing_process_is_not_timeout(tmp_path): + command = [sys.executable, "-c", ( + "from pathlib import Path; import time; Path('started').touch(); " + "print('provider started', flush=True); time.sleep(5)" + )] + timed_out = run_command(command, execute=True, cwd=tmp_path, timeout_seconds=0.3) + assert (tmp_path / "started").exists() + assert timed_out["ok"] is False + assert timed_out["executed"] is True + assert timed_out["returncode"] is None + assert timed_out["timed_out"] is True + assert "provider started" in timed_out["stdout"] + missing = run_command([str(tmp_path / "missing-executable")], execute=True) + assert missing["ok"] is False + assert missing["returncode"] is None + assert missing.get("timed_out", False) is False + assert missing["stderr"] + + +@pytest.mark.parametrize(("result", "expected"), [ + ({"json": {"error": {"message": "provider error"}}, "stderr": "fallback"}, "provider error"), + ({"json": {"error": "denied"}}, "denied"), + ({"json": {"code": 42, "msg": "rejected"}, "stderr": "fallback"}, "rejected"), + ({"json": {"code": 0, "msg": "not an error"}, "stderr": " failure\n details "}, "failure details"), + ({"stdout": " fallback\n output "}, "fallback output"), + ({"stderr": "x" * 1900}, "x" * 1800), +]) +def test_provider_error_precedence_and_bounded_fallback(result, expected): + assert command_error(result) == expected + + +def test_channel_transport_runs_without_presentation_or_control_plane_sources(tmp_path): + source = Path(__file__).resolve().parents[2] / "loopx" + package = tmp_path / "loopx" + for relative in ( + "__init__.py", "extensions/__init__.py", "extensions/lark/__init__.py", + "extensions/lark/cli_transport.py", "extensions/lark/goal_channel_transport.py", + "extensions/lark/identity_shapes.py", + ): + destination = package / relative + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source / relative, destination) + assert not (package / "extensions/lark/presentation").exists() + assert not (package / "control_plane").exists() + code = """ +import sys +from pathlib import Path +sys.path.insert(0, sys.argv[1]) +from loopx.extensions.lark import cli_transport, goal_channel_transport +assert Path(cli_transport.__file__).is_relative_to(Path(sys.argv[1])) +result = goal_channel_transport.call(cli_transport.default_subprocess_runner, + [sys.executable, '-c', 'import json; print(json.dumps(dict(ok=True, message="complete")))']) +assert result['returncode'] == 0 +assert goal_channel_transport.json_payload(result) == {'ok': True, 'message': 'complete'} +assert not any(name.startswith(('loopx.control_plane', 'loopx.extensions.lark.presentation')) + for name in sys.modules) +""" + completed = subprocess.run( + [sys.executable, "-I", "-c", code, str(tmp_path)], + cwd=tmp_path, capture_output=True, text=True, encoding="utf-8", timeout=10, + ) + assert completed.returncode == 0, completed.stderr diff --git a/tests/test_explore_presentation_views.py b/tests/test_explore_presentation_views.py index 325185aed1..4a0b53c330 100644 --- a/tests/test_explore_presentation_views.py +++ b/tests/test_explore_presentation_views.py @@ -998,7 +998,7 @@ def fail_runner(command, *_args, **_kwargs): return {"ok": False, "returncode": 1, "stderr": "should not be invoked"} monkeypatch.setattr( - "loopx.extensions.lark.presentation.explore_results._run_command", + "loopx.extensions.lark.presentation.explore_results.run_command", fail_runner, )