Coverage for src / apcore_cli / strategy.py: 84%
83 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-04-26 10:23 +0800
1"""Pipeline strategy commands — describe-pipeline (FE-11)."""
3from __future__ import annotations
5import contextlib
6import json
7import logging
8import sys
9from typing import Any
11import click
13from apcore_cli.output import resolve_format
15logger = logging.getLogger(__name__)
17_PRESET_STEPS = {
18 "standard": [
19 "context_creation",
20 "call_chain_guard",
21 "module_lookup",
22 "acl_check",
23 "approval_gate",
24 "middleware_before",
25 "input_validation",
26 "execute",
27 "output_validation",
28 "middleware_after",
29 "return_result",
30 ],
31 "internal": [
32 "context_creation",
33 "call_chain_guard",
34 "module_lookup",
35 "middleware_before",
36 "input_validation",
37 "execute",
38 "output_validation",
39 "middleware_after",
40 "return_result",
41 ],
42 "testing": [
43 "context_creation",
44 "module_lookup",
45 "middleware_before",
46 "input_validation",
47 "execute",
48 "output_validation",
49 "middleware_after",
50 "return_result",
51 ],
52 "performance": [
53 "context_creation",
54 "call_chain_guard",
55 "module_lookup",
56 "acl_check",
57 "approval_gate",
58 "input_validation",
59 "execute",
60 "output_validation",
61 "return_result",
62 ],
63 "minimal": [
64 "context_creation",
65 "module_lookup",
66 "execute",
67 "return_result",
68 ],
69}
72def _render_pipeline_table(
73 steps_info: list[dict[str, Any]],
74 fmt: str,
75 strategy_name: str,
76 step_count: int,
77) -> None:
78 """Render pipeline steps as JSON or a table.
80 When fmt == "json" (or stdout is not a TTY), emits JSON.
81 Otherwise renders a full-metadata table (pure/removable/timeout columns) when
82 any step carries non-default metadata; renders a name-only table otherwise.
83 """
84 if fmt == "json" or not sys.stdout.isatty():
85 payload = {
86 "strategy": strategy_name,
87 "step_count": step_count,
88 "steps": [{"index": i + 1, **s} for i, s in enumerate(steps_info)],
89 }
90 click.echo(json.dumps(payload, indent=2))
91 return
93 click.echo(f"Pipeline: {strategy_name} ({step_count} steps)\n")
94 has_metadata = steps_info and any(
95 s.get("pure") is not False or s.get("removable") is not True or s.get("timeout_ms") for s in steps_info
96 )
98 if has_metadata:
99 click.echo(f" {'#':<4} {'Step':<28} {'Pure':<6} {'Removable':<11} Timeout")
100 click.echo(f" {'-' * 4} {'-' * 28} {'-' * 6} {'-' * 11} {'-' * 8}")
101 for i, s in enumerate(steps_info, 1):
102 pure = "yes" if s.get("pure") else "no"
103 removable = "yes" if s.get("removable", True) else "no"
104 timeout = f"{s['timeout_ms']}ms" if s.get("timeout_ms") else "\u2014"
105 click.echo(f" {i:<4} {s['name']:<28} {pure:<6} {removable:<11} {timeout}")
106 else:
107 click.echo(f" {'#':<4} {'Step':<28}")
108 click.echo(f" {'-' * 4} {'-' * 28}")
109 for i, s in enumerate(steps_info, 1):
110 click.echo(f" {i:<4} {s['name']:<28}")
113def register_pipeline_command(cli: click.Group, executor: Any) -> None:
114 """Register the describe-pipeline command."""
116 @cli.command("describe-pipeline")
117 @click.option(
118 "--strategy",
119 type=click.Choice(["standard", "internal", "testing", "performance", "minimal"]),
120 default="standard",
121 help="Strategy to describe (default: standard).",
122 )
123 @click.option("--format", "output_format", type=click.Choice(["table", "json"]), default=None)
124 def _describe_pipeline_cmd(strategy: str, output_format: str | None) -> None: # pyright: ignore[reportUnusedVariable]
125 """Show the execution pipeline steps for a strategy."""
126 fmt = resolve_format(output_format)
128 # Try to get StrategyInfo from executor.describe_pipeline() (apcore >= 0.18.0)
129 strategy_info = None
130 if hasattr(executor, "describe_pipeline"):
131 with contextlib.suppress(AttributeError, NotImplementedError, TypeError):
132 strategy_info = executor.describe_pipeline(strategy)
134 if strategy_info is not None:
135 # Use StrategyInfo dataclass fields for output
136 info_name = getattr(strategy_info, "name", strategy)
137 info_step_count = getattr(strategy_info, "step_count", 0)
138 info_step_names = getattr(strategy_info, "step_names", [])
140 # Try to get full step metadata from executor._strategy.steps
141 steps_info: list[dict[str, Any]] = []
142 strategy_obj = None
143 if hasattr(executor, "_strategy"):
144 try:
145 strategy_obj = executor._strategy
146 except (AttributeError, NotImplementedError, TypeError) as e:
147 logger.warning("executor._strategy access failed; using step-names fallback: %s", e)
148 if strategy_obj is None and hasattr(executor, "_resolve_strategy_name"):
149 try:
150 strategy_obj = executor._resolve_strategy_name(strategy)
151 except (AttributeError, NotImplementedError, TypeError) as e:
152 logger.warning("executor._resolve_strategy_name failed; using step-names fallback: %s", e)
154 if strategy_obj is not None and hasattr(strategy_obj, "steps"):
155 for step in strategy_obj.steps:
156 steps_info.append(
157 {
158 "name": getattr(step, "name", ""),
159 "pure": getattr(step, "pure", False),
160 "removable": getattr(step, "removable", True),
161 "timeout_ms": getattr(step, "timeout_ms", None),
162 }
163 )
164 else:
165 steps_info = [
166 {"name": s, "pure": False, "removable": True, "timeout_ms": None} for s in info_step_names
167 ]
169 _render_pipeline_table(steps_info, fmt, info_name, info_step_count)
170 return
172 # Fall back to legacy _resolve_strategy_name (apcore < 0.18.0)
173 strategy_obj = None
174 if hasattr(executor, "_resolve_strategy_name"):
175 try:
176 strategy_obj = executor._resolve_strategy_name(strategy)
177 except (AttributeError, NotImplementedError, TypeError) as e:
178 logger.warning("executor._resolve_strategy_name failed; using preset-steps fallback: %s", e)
180 if strategy_obj is None:
181 # Provide static info for known strategies
182 steps = _PRESET_STEPS.get(strategy, [])
184 if fmt == "json" or not sys.stdout.isatty():
185 payload = {
186 "strategy": strategy,
187 "step_count": len(steps),
188 "steps": [{"index": i + 1, "name": s} for i, s in enumerate(steps)],
189 }
190 click.echo(json.dumps(payload, indent=2))
191 else:
192 click.echo(f"Pipeline: {strategy} ({len(steps)} steps)\n")
193 click.echo(f" {'#':<4} {'Step':<28}")
194 click.echo(f" {'-' * 4} {'-' * 28}")
195 for i, s in enumerate(steps, 1):
196 click.echo(f" {i:<4} {s:<28}")
197 return
199 # Use actual strategy object for detailed info
200 steps_info = []
201 for step in strategy_obj.steps:
202 step_entry: dict[str, Any] = {
203 "name": step.name,
204 "pure": getattr(step, "pure", False),
205 "removable": getattr(step, "removable", True),
206 "timeout_ms": getattr(step, "timeout_ms", None),
207 }
208 steps_info.append(step_entry)
210 _render_pipeline_table(steps_info, fmt, strategy, len(steps_info))