feat(acms): context analysis produces meaningful summaries

Add phase-wise context analysis summaries for strategize, execute, and apply, and integrate them into `agents project context inspect` and `agents project context simulate` with per-phase resource counts, size and token utilization, and narrowing diagnostics.\n\nAdd Behave and Robot coverage for empty, single-resource, multi-resource, and budget-constrained context analysis, and tighten the M5 smoke coverage after review feedback with exact summary assertions and a missing-path regression scenario for `agents context add`.\n\nISSUES CLOSED: #849

CHANGELOG.md

@@ -214,6 +214,12 @@
   workspace-symbols query schema, and defensive schema validation.
   Extended `initialize()` to advertise all 11 capabilities. Fixed
   `workspace_symbols` runtime handler to accept query-only input. (#834)
+- Added per-phase ACMS context analysis summaries for
+  `agents project context inspect/simulate` with human-readable metrics
+  (fragment/resource counts, size/tokens, budget utilization), explicit
+  strategize->execute->apply narrowing diagnostics, Robot acceptance
+  verification upgrades, and Behave edge-case coverage for empty,
+  single-resource, multi-resource, and budget-constrained contexts. (#849)
 - Added TDD bug-capture E2E tests for bug #1028 — ACMS indexing pipeline not
   wired into CLI. Four Robot Framework E2E tests prove ContextTierService starts
   empty on every CLI invocation. Tests use `@tdd_expected_fail` until the bug

features/m5_acms_smoke.feature

@@ -96,6 +96,12 @@ Feature: M5 ACMS pipeline and large-project context smoke tests
     When I m5 smoke invoke context add with path "src/main.py"
     Then the m5 smoke context add should succeed
 
+  Scenario: M5 smoke context add missing path reports an error
+    Given a m5 smoke project with mocked context service
+    When I m5 smoke invoke context add with missing path "src/missing.py"
+    Then the m5 smoke context add output should report the missing path
+    And the m5 smoke context service should not add any files
+
   Scenario: M5 smoke context show displays file content
     Given a m5 smoke project with context entries
     When I m5 smoke invoke context show for path "src/main.py"
@@ -124,7 +130,7 @@ Feature: M5 ACMS pipeline and large-project context smoke tests
     Given a m5 smoke project with multiple context entries
     When I m5 smoke invoke context show without path
     Then the m5 smoke context show should succeed
-    And the m5 smoke context show output should contain summary
+    And the m5 smoke context show output should contain an exact summary
 
   # --- Context CLI: clear and re-add ---
 

features/project_context_phase_analysis.feature

@@ -0,0 +1,31 @@
+Feature: Project context phase analysis summaries
+  As a developer
+  I want per-phase ACMS context analysis summaries
+  So that strategize/execute/apply scope narrowing is visible and actionable
+
+  Scenario: Empty context still produces per-phase summaries
+    Given a phase analysis policy with narrowing limits
+    And no tiered fragments for phase analysis
+    When I compute project context phase analysis with budget 2000
+    Then each phase summary should include resource and size metrics
+    And phase analysis should report monotonic narrowing
+
+  Scenario: Single resource context produces human-readable summaries
+    Given a phase analysis policy with narrowing limits
+    And a single resource fragment for phase analysis
+    When I compute project context phase analysis with budget 2000
+    Then each phase summary should include human-readable text
+    And strategize phase should include at least one fragment
+
+  Scenario: Multiple resources are counted in summaries
+    Given a phase analysis policy with narrowing limits
+    And multiple resource fragments for phase analysis
+    When I compute project context phase analysis with budget 2000
+    Then strategize phase resource count should be 2
+
+  Scenario: Budget constraints reduce execute and apply scope
+    Given a phase analysis policy with narrowing limits
+    And budget constrained fragments for phase analysis
+    When I compute project context phase analysis with budget 1000
+    Then execute phase should have fewer tokens than strategize phase
+    And apply phase should have fewer or equal tokens than execute phase

features/steps/m5_acms_smoke_steps.py

@@ -347,6 +347,20 @@ def step_m5_invoke_context_add(context: Context, path: str) -> None:
         context.m5_result = result
 
 
+@when('I m5 smoke invoke context add with missing path "{path}"')
+def step_m5_invoke_context_add_missing_path(context: Context, path: str) -> None:
+    container = _mock_container(context.mock_context_service)
+    with (
+        patch(
+            "cleveragents.application.container.get_container",
+            return_value=container,
+        ),
+        patch.object(Path, "exists", return_value=False),
+    ):
+        result = context.runner.invoke(context_app, ["add", path])
+        context.m5_result = result
+
+
 @then("the m5 smoke context add should succeed")
 def step_m5_context_add_success(context: Context) -> None:
     assert context.m5_result.exit_code == 0, (
@@ -354,6 +368,20 @@ def step_m5_context_add_success(context: Context) -> None:
     )
 
 
+@then("the m5 smoke context add output should report the missing path")
+def step_m5_context_add_missing_path_reported(context: Context) -> None:
+    output = context.m5_result.output
+    assert "Path does not exist:" in output, f"Missing-path error not shown: {output}"
+    assert "No files were added to context." in output, (
+        f"Expected add command to report no-op after missing path: {output}"
+    )
+
+
+@then("the m5 smoke context service should not add any files")
+def step_m5_context_service_not_called(context: Context) -> None:
+    context.mock_context_service.add_to_context.assert_not_called()
+
+
 @when('I m5 smoke invoke context show for path "{path}"')
 def step_m5_invoke_context_show(context: Context, path: str) -> None:
     container = _mock_container(context.mock_context_service)
@@ -480,6 +508,10 @@ def step_m5_invoke_context_inspect(context: Context) -> None:
             "cleveragents.application.container.get_container",
             return_value=mock_container,
         ),
+        patch(
+            "cleveragents.cli.commands.project_context._read_policy",
+            return_value=ProjectContextPolicy(),
+        ),
         patch(
             "cleveragents.cli.commands.project_context._read_acms_config",
             return_value=_default_acms_config(),
@@ -512,6 +544,10 @@ def step_m5_invoke_context_simulate(context: Context) -> None:
             "cleveragents.application.container.get_container",
             return_value=mock_container,
         ),
+        patch(
+            "cleveragents.cli.commands.project_context._read_policy",
+            return_value=ProjectContextPolicy(),
+        ),
         patch(
             "cleveragents.cli.commands.project_context._read_acms_config",
             return_value=_default_acms_config(),
@@ -643,14 +679,17 @@ def step_m5_invoke_context_show_no_path(context: Context) -> None:
         context.m5_result = result
 
 
-@then("the m5 smoke context show output should contain summary")
+@then("the m5 smoke context show output should contain an exact summary")
 def step_m5_context_show_summary(context: Context) -> None:
     output = context.m5_result.output
+    assert "Context Summary:" in output, f"Expected summary header in output: {output}"
+    assert "Total files: 3" in output, f"Expected file count in output: {output}"
+    assert "Total size: 6,144 bytes" in output, (
+        f"Expected aggregate size in output: {output}"
+    )
     assert (
-        "total" in output.lower()
-        or "summary" in output.lower()
-        or "files" in output.lower()
-    ), f"Expected summary in output, got: {output}"
+        "Use 'agents context show <file>' to view specific file content." in output
+    ), f"Expected usage hint in output: {output}"
 
 
 # ---------------------------------------------------------------------------

features/steps/project_context_phase_analysis_steps.py

@@ -0,0 +1,185 @@
+"""Behave steps for project_context_phase_analysis.feature."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from behave import given, then, when
+
+from cleveragents.application.services.context_phase_analysis import (
+    analyze_phase_summaries,
+)
+from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment
+from cleveragents.domain.models.core.context_policy import (
+    ContextView,
+    ProjectContextPolicy,
+)
+
+
+def _base_policy() -> ProjectContextPolicy:
+    return ProjectContextPolicy(
+        default_view=ContextView(
+            include_resources=["local/*"],
+            max_file_size=2_000_000,
+            max_total_size=5_000_000,
+        ),
+        strategize_view=ContextView(
+            include_resources=["local/*"],
+            include_paths=["src/**/*.py", "docs/**/*.md"],
+            exclude_paths=["**/test_*.py"],
+            max_file_size=1_000_000,
+            max_total_size=2_000_000,
+        ),
+        execute_view=ContextView(
+            include_resources=["local/*"],
+            include_paths=["src/**/*.py"],
+            exclude_paths=["**/test_*.py", "**/conftest.py"],
+            max_file_size=500_000,
+            max_total_size=700_000,
+        ),
+        apply_view=ContextView(
+            include_resources=["local/*"],
+            include_paths=["src/**/*.py"],
+            exclude_paths=["docs/**"],
+            max_file_size=250_000,
+            max_total_size=400_000,
+        ),
+    )
+
+
+@given("a phase analysis policy with narrowing limits")
+def step_policy(context: Any) -> None:
+    context.phase_policy = _base_policy()
+
+
+@given("no tiered fragments for phase analysis")
+def step_no_fragments(context: Any) -> None:
+    context.phase_fragments = []
+
+
+@given("a single resource fragment for phase analysis")
+def step_single_fragment(context: Any) -> None:
+    context.phase_fragments = [
+        TieredFragment(
+            fragment_id="single",
+            content="x" * 100,
+            tier=ContextTier.HOT,
+            resource_id="local/repo-a",
+            project_name="local/ctx-app",
+            token_count=120,
+            metadata={"path": "src/main.py", "byte_size": 100_000},
+        )
+    ]
+
+
+@given("multiple resource fragments for phase analysis")
+def step_multiple_resources(context: Any) -> None:
+    context.phase_fragments = [
+        TieredFragment(
+            fragment_id="a",
+            content="x" * 80,
+            tier=ContextTier.HOT,
+            resource_id="local/repo-a",
+            project_name="local/ctx-app",
+            token_count=100,
+            metadata={"path": "src/a.py", "byte_size": 80_000},
+        ),
+        TieredFragment(
+            fragment_id="b",
+            content="x" * 90,
+            tier=ContextTier.WARM,
+            resource_id="local/repo-b",
+            project_name="local/ctx-app",
+            token_count=110,
+            metadata={"path": "src/b.py", "byte_size": 90_000},
+        ),
+    ]
+
+
+@given("budget constrained fragments for phase analysis")
+def step_budget_fragments(context: Any) -> None:
+    context.phase_fragments = [
+        TieredFragment(
+            fragment_id="f1",
+            content="x" * 400,
+            tier=ContextTier.HOT,
+            resource_id="local/repo-a",
+            project_name="local/ctx-app",
+            token_count=450,
+            metadata={"path": "src/core.py", "byte_size": 400_000},
+        ),
+        TieredFragment(
+            fragment_id="f2",
+            content="x" * 380,
+            tier=ContextTier.WARM,
+            resource_id="local/repo-a",
+            project_name="local/ctx-app",
+            token_count=430,
+            metadata={"path": "src/engine.py", "byte_size": 380_000},
+        ),
+        TieredFragment(
+            fragment_id="f3",
+            content="x" * 200,
+            tier=ContextTier.COLD,
+            resource_id="local/repo-a",
+            project_name="local/ctx-app",
+            token_count=220,
+            metadata={"path": "docs/design.md", "byte_size": 200_000},
+        ),
+    ]
+
+
+@when("I compute project context phase analysis with budget {budget:d}")
+def step_compute(context: Any, budget: int) -> None:
+    context.phase_result = analyze_phase_summaries(
+        context.phase_policy,
+        context.phase_fragments,
+        budget,
+    )
+
+
+@then("each phase summary should include resource and size metrics")
+def step_metrics_present(context: Any) -> None:
+    for phase in ("strategize", "execute", "apply"):
+        row = context.phase_result["phases"][phase]
+        assert "resource_count" in row
+        assert "total_bytes" in row
+        assert "budget_used_ratio" in row
+
+
+@then("phase analysis should report monotonic narrowing")
+def step_narrowing(context: Any) -> None:
+    narrowing = context.phase_result["narrowing"]
+    assert narrowing["overall"] is True
+
+
+@then("each phase summary should include human-readable text")
+def step_summary_text(context: Any) -> None:
+    for phase in ("strategize", "execute", "apply"):
+        summary = context.phase_result["phases"][phase]["summary"]
+        assert isinstance(summary, str)
+        assert phase in summary
+
+
+@then("strategize phase should include at least one fragment")
+def step_strat_non_empty(context: Any) -> None:
+    assert context.phase_result["phases"]["strategize"]["fragment_count"] >= 1
+
+
+@then("strategize phase resource count should be 2")
+def step_resource_count_two(context: Any) -> None:
+    assert context.phase_result["phases"]["strategize"]["resource_count"] == 2
+
+
+@then("execute phase should have fewer tokens than strategize phase")
+def step_execute_less(context: Any) -> None:
+    strat_tokens = context.phase_result["phases"]["strategize"]["total_tokens"]
+    exec_tokens = context.phase_result["phases"]["execute"]["total_tokens"]
+    assert exec_tokens < strat_tokens
+
+
+@then("apply phase should have fewer or equal tokens than execute phase")
+def step_apply_less_or_equal(context: Any) -> None:
+    exec_tokens = context.phase_result["phases"]["execute"]["total_tokens"]
+    apply_tokens = context.phase_result["phases"]["apply"]["total_tokens"]
+    assert apply_tokens <= exec_tokens

robot/helper_m5_e2e_verification.py

@@ -23,16 +23,19 @@ import sys
 import tempfile
 from collections.abc import Callable
 from pathlib import Path
-from typing import Any, NoReturn
+from typing import Any, NoReturn, cast
 
 from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import Session, sessionmaker
 
 # Ensure src is importable when run from workspace root
 _SRC = str(Path(__file__).resolve().parents[1] / "src")
 if _SRC not in sys.path:
     sys.path.insert(0, _SRC)
 
+from cleveragents.application.services.context_phase_analysis import (  # noqa: E402
+    analyze_phase_summaries,
+)
 from cleveragents.application.services.resource_registry_service import (  # noqa: E402
     ResourceRegistryService,
 )
@@ -40,6 +43,10 @@ from cleveragents.cli.commands.project_context import (  # noqa: E402
     _read_policy,
     _write_policy,
 )
+from cleveragents.domain.models.acms.tiers import (  # noqa: E402
+    ContextTier,
+    TieredFragment,
+)
 from cleveragents.domain.models.core.context_policy import (  # noqa: E402
     VALID_PHASES,
     ContextView,
@@ -76,7 +83,7 @@ class _NoClose:
     def close(self) -> None:
         pass
 
-    def __getattr__(self, name: str) -> object:
+    def __getattr__(self, name: str) -> Any:
         return getattr(object.__getattribute__(self, "_s"), name)
 
 
@@ -96,8 +103,8 @@ def _setup_db() -> tuple[
     session = sessionmaker(bind=engine, expire_on_commit=False)()
     wrapper = _NoClose(session)
 
-    def factory() -> object:
-        return wrapper
+    def factory() -> Session:
+        return cast(Session, wrapper)
 
     proj_repo = NamespacedProjectRepository(session_factory=factory)
     link_repo = ProjectResourceLinkRepository(session_factory=factory)
@@ -512,14 +519,14 @@ def acms_scoped_context() -> None:
             include_paths=["src/**/*.py"],
             exclude_paths=["**/test_*.py", "**/conftest.py"],
             max_file_size=500_000,
-            max_total_size=50_000_000,
+            max_total_size=1_200_000,
         ),
         apply_view=ContextView(
             include_resources=["local/large-repo"],
             include_paths=["src/**/*.py"],
-            exclude_paths=["docs/**"],
+            exclude_paths=["docs/**", "**/test_*.py", "**/conftest.py"],
             max_file_size=250_000,
-            max_total_size=25_000_000,
+            max_total_size=700_000,
         ),
     )
 
@@ -556,6 +563,76 @@ def acms_scoped_context() -> None:
             f"{apply_view.max_file_size}"
         )
 
+    # Verify per-phase analysis summaries and narrowing diagnostics
+    sample_fragments = [
+        TieredFragment(
+            fragment_id="f-core",
+            content="x" * 900,
+            tier=ContextTier.HOT,
+            resource_id="local/large-repo",
+            project_name="local/scoped-test",
+            token_count=900,
+            metadata={"path": "src/core.py", "byte_size": 900_000},
+        ),
+        TieredFragment(
+            fragment_id="f-engine",
+            content="x" * 500,
+            tier=ContextTier.HOT,
+            resource_id="local/large-repo",
+            project_name="local/scoped-test",
+            token_count=500,
+            metadata={"path": "src/engine.py", "byte_size": 450_000},
+        ),
+        TieredFragment(
+            fragment_id="f-test",
+            content="x" * 120,
+            tier=ContextTier.WARM,
+            resource_id="local/large-repo",
+            project_name="local/scoped-test",
+            token_count=120,
+            metadata={"path": "src/test_core.py", "byte_size": 100_000},
+        ),
+        TieredFragment(
+            fragment_id="f-doc",
+            content="x" * 180,
+            tier=ContextTier.WARM,
+            resource_id="local/large-repo",
+            project_name="local/scoped-test",
+            token_count=180,
+            metadata={"path": "docs/design.md", "byte_size": 200_000},
+        ),
+        TieredFragment(
+            fragment_id="f-conftest",
+            content="x" * 90,
+            tier=ContextTier.COLD,
+            resource_id="local/large-repo",
+            project_name="local/scoped-test",
+            token_count=90,
+            metadata={"path": "src/conftest.py", "byte_size": 80_000},
+        ),
+        TieredFragment(
+            fragment_id="f-huge",
+            content="x" * 1500,
+            tier=ContextTier.COLD,
+            resource_id="local/large-repo",
+            project_name="local/scoped-test",
+            token_count=1500,
+            metadata={"path": "src/huge.py", "byte_size": 1_500_000},
+        ),
+    ]
+    analysis = analyze_phase_summaries(policy, sample_fragments, budget_tokens=2_000)
+
+    for phase in ("strategize", "execute", "apply"):
+        phase_row = analysis["phases"][phase]
+        if not phase_row["summary"]:
+            _fail(f"{phase} summary should be non-empty")
+        if "resource_count" not in phase_row or "total_bytes" not in phase_row:
+            _fail(f"{phase} summary missing resource/size metrics")
+
+    narrowing = analysis["narrowing"]
+    if not narrowing["overall"]:
+        _fail(f"narrowing should hold across phases: {narrowing}")
+
     # Verify persistence round-trip
     proj_repo, _link_repo, _svc, sf = _setup_db()
     _create_project(proj_repo, "local/scoped-test")

src/cleveragents/application/services/context_phase_analysis.py

@@ -0,0 +1,165 @@
+"""Phase-wise context analysis summaries for ACMS project context views.
+
+Builds per-phase summaries (strategize/execute/apply) from tiered fragments
+using the project's resolved ``ProjectContextPolicy``. The summaries are
+designed to be human-readable and machine-assertable for CLI output and tests.
+"""
+
+from __future__ import annotations
+
+import fnmatch
+from pathlib import PurePosixPath
+from typing import Any
+
+from cleveragents.domain.models.acms.tiers import TieredFragment
+from cleveragents.domain.models.core.context_policy import ProjectContextPolicy
+
+
+def _resource_matches(resource_id: str, include_patterns: list[str]) -> bool:
+    """Return ``True`` when the resource is included by patterns."""
+    if not include_patterns:
+        return True
+    return any(fnmatch.fnmatch(resource_id, pattern) for pattern in include_patterns)
+
+
+def _path_matches(path: str, include_patterns: list[str]) -> bool:
+    """Return ``True`` when the path is included by patterns."""
+    if not include_patterns:
+        return True
+    path_obj = PurePosixPath(path)
+    return any(_matches_pattern(path_obj, pattern) for pattern in include_patterns)
+
+
+def _matches_pattern(path_obj: PurePosixPath, pattern: str) -> bool:
+    """Match with a small compatibility shim for ``**/`` zero-depth cases."""
+    return path_obj.match(pattern) or (
+        "**/" in pattern and path_obj.match(pattern.replace("**/", ""))
+    )
+
+
+def _extract_path(fragment: TieredFragment) -> str:
+    """Extract a best-effort source path for include/exclude path filtering."""
+    raw_path = fragment.metadata.get("path")
+    if isinstance(raw_path, str) and raw_path:
+        return raw_path
+    return fragment.resource_id or ""
+
+
+def _fragment_size_bytes(fragment: TieredFragment) -> int:
+    """Resolve fragment size in bytes with metadata override when present."""
+    raw_size = fragment.metadata.get("byte_size")
+    if isinstance(raw_size, int) and raw_size >= 0:
+        return raw_size
+    return len(fragment.content.encode("utf-8"))
+
+
+def analyze_phase_summaries(
+    policy: ProjectContextPolicy,
+    fragments: list[TieredFragment],
+    budget_tokens: int,
+) -> dict[str, Any]:
+    """Build per-phase summaries and narrowing diagnostics.
+
+    Returns a dictionary with:
+    - ``phases``: per-phase summary rows
+    - ``narrowing``: monotonic checks from strategize->execute->apply
+    """
+
+    phase_names = ["strategize", "execute", "apply"]
+    rows: dict[str, dict[str, Any]] = {}
+
+    for phase in phase_names:
+        view = policy.resolve_view(phase)
+        selected: list[TieredFragment] = []
+
+        excluded_by_resource = 0
+        excluded_by_path = 0
+        excluded_by_file_size = 0
+        excluded_by_total_size = 0
+
+        total_bytes = 0
+        total_tokens = 0
+
+        for fragment in fragments:
+            resource_id = fragment.resource_id or ""
+            path = _extract_path(fragment)
+
+            if not _resource_matches(resource_id, view.include_resources):
+                excluded_by_resource += 1
+                continue
+            if any(fnmatch.fnmatch(resource_id, p) for p in view.exclude_resources):
+                excluded_by_resource += 1
+                continue
+
+            if not _path_matches(path, view.include_paths):
+                excluded_by_path += 1
+                continue
+            path_obj = PurePosixPath(path)
+            if any(_matches_pattern(path_obj, p) for p in view.exclude_paths):
+                excluded_by_path += 1
+                continue
+
+            file_size = _fragment_size_bytes(fragment)
+            if view.max_file_size is not None and file_size > view.max_file_size:
+                excluded_by_file_size += 1
+                continue
+
+            if (
+                view.max_total_size is not None
+                and total_bytes + file_size > view.max_total_size
+            ):
+                excluded_by_total_size += 1
+                continue
+
+            selected.append(fragment)
+            total_bytes += file_size
+            total_tokens += fragment.token_count
+
+        distinct_resources = len({f.resource_id for f in selected if f.resource_id})
+        budget_used = (total_tokens / budget_tokens) if budget_tokens > 0 else 0.0
+
+        rows[phase] = {
+            "phase": phase,
+            "fragment_count": len(selected),
+            "resource_count": distinct_resources,
+            "total_bytes": total_bytes,
+            "total_tokens": total_tokens,
+            "budget_tokens": budget_tokens,
+            "budget_used_ratio": min(max(budget_used, 0.0), 1.0),
+            "excluded": {
+                "resource_or_scope": excluded_by_resource,
+                "path_rules": excluded_by_path,
+                "max_file_size": excluded_by_file_size,
+                "max_total_size": excluded_by_total_size,
+            },
+            "summary": (
+                f"{phase}: {len(selected)} fragments from {distinct_resources} "
+                f"resources, {total_bytes} bytes, {total_tokens} tokens "
+                f"({min(max(budget_used, 0.0), 1.0):.1%} budget)"
+            ),
+        }
+
+    strat = rows["strategize"]
+    exe = rows["execute"]
+    app = rows["apply"]
+
+    tokens_non_increasing = (
+        strat["total_tokens"] >= exe["total_tokens"] >= app["total_tokens"]
+    )
+    bytes_non_increasing = (
+        strat["total_bytes"] >= exe["total_bytes"] >= app["total_bytes"]
+    )
+    fragments_non_increasing = (
+        strat["fragment_count"] >= exe["fragment_count"] >= app["fragment_count"]
+    )
+
+    narrowing = {
+        "tokens_non_increasing": tokens_non_increasing,
+        "bytes_non_increasing": bytes_non_increasing,
+        "fragments_non_increasing": fragments_non_increasing,
+        "overall": (
+            tokens_non_increasing and bytes_non_increasing and fragments_non_increasing
+        ),
+    }
+
+    return {"phases": rows, "narrowing": narrowing}

src/cleveragents/cli/commands/project_context.py

@@ -30,6 +30,9 @@ from rich.console import Console
 from rich.panel import Panel
 from rich.table import Table
 
+from cleveragents.application.services.context_phase_analysis import (
+    analyze_phase_summaries,
+)
 from cleveragents.cli.formatting import OutputFormat, format_output
 from cleveragents.domain.models.acms.crp import (
     AssembledContext,
@@ -901,6 +904,7 @@ def context_inspect(
 
     # Read ACMS config and apply overrides
     acms = _read_acms_config(session_factory, project)
+    policy = _read_policy(session_factory, project)
     if breadth is not None:
         acms["default_breadth"] = breadth
     if depth is not None:
@@ -909,6 +913,9 @@ def context_inspect(
         except ValueError:
             acms["default_depth"] = depth
 
+    budget_tokens = int(acms.get("hot_max_tokens", _DEFAULT_HOT_MAX_TOKENS))
+    phase_analysis = analyze_phase_summaries(policy, project_fragments, budget_tokens)
+
     # Build result
     data: dict[str, Any] = {
         "project": project,
@@ -933,6 +940,7 @@ def context_inspect(
         },
         "actor_visibility": actor_views,
         "acms_config": acms,
+        "phase_analysis": phase_analysis,
     }
 
     # Add fragment details (limited to first 20)
@@ -986,6 +994,53 @@ def context_inspect(
         console.print(
             Panel("\n".join(av_lines), title="Actor Visibility", expand=False)
         )
+
+        phase_table = Table(title="Per-Phase Context Analysis", expand=False)
+        phase_table.add_column("Phase", style="bold")
+        phase_table.add_column("Fragments", justify="right")
+        phase_table.add_column("Resources", justify="right")
+        phase_table.add_column("Bytes", justify="right")
+        phase_table.add_column("Tokens", justify="right")
+        phase_table.add_column("Budget", justify="right")
+        for phase in ["strategize", "execute", "apply"]:
+            row = phase_analysis["phases"][phase]
+            phase_table.add_row(
+                phase,
+                str(row["fragment_count"]),
+                str(row["resource_count"]),
+                str(row["total_bytes"]),
+                str(row["total_tokens"]),
+                f"{row['budget_used_ratio']:.1%}",
+            )
+        console.print(phase_table)
+
+        narrowing = phase_analysis["narrowing"]
+        console.print(
+            Panel(
+                "\n".join(
+                    [
+                        "[bold]Narrowing checks:[/bold]",
+                        (
+                            "tokens strategize >= execute >= apply: "
+                            f"{narrowing['tokens_non_increasing']}"
+                        ),
+                        (
+                            "bytes strategize >= execute >= apply: "
+                            f"{narrowing['bytes_non_increasing']}"
+                        ),
+                        (
+                            "fragments strategize >= execute >= apply: "
+                            f"{narrowing['fragments_non_increasing']}"
+                        ),
+                    ]
+                ),
+                title=(
+                    "Phase Narrowing "
+                    f"({'OK' if narrowing['overall'] else 'Needs Review'})"
+                ),
+                expand=False,
+            )
+        )
     else:
         console.print(format_output(data, output_format))
 
@@ -1057,6 +1112,11 @@ def context_simulate(
         raise typer.Exit(1) from exc
 
     acms = _read_acms_config(session_factory, project)
+    policy = _read_policy(session_factory, project)
+    tier_service = _get_context_tier_service()
+    project_fragments = tier_service.get_scoped_view([project])
+    budget_tokens = int(acms.get("hot_max_tokens", _DEFAULT_HOT_MAX_TOKENS))
+    phase_analysis = analyze_phase_summaries(policy, project_fragments, budget_tokens)
     assembled = _simulate_context_assembly(
         project_name=project,
         acms_config=acms,
@@ -1070,6 +1130,7 @@ def context_simulate(
     data["project"] = project
     data["view"] = effective_view
     data["acms_config"] = acms
+    data["phase_analysis"] = phase_analysis
 
     if output_format.lower() == OutputFormat.RICH:
         title = f"Context Simulation: {project} (view: {effective_view})"
@@ -1107,5 +1168,24 @@ def context_simulate(
                     frag.provenance.strategy,
                 )
             console.print(frag_table)
+
+        phase_table = Table(title="Per-Phase Context Analysis", expand=False)
+        phase_table.add_column("Phase", style="bold")
+        phase_table.add_column("Fragments", justify="right")
+        phase_table.add_column("Resources", justify="right")
+        phase_table.add_column("Bytes", justify="right")
+        phase_table.add_column("Tokens", justify="right")
+        phase_table.add_column("Budget", justify="right")
+        for phase in ["strategize", "execute", "apply"]:
+            row = phase_analysis["phases"][phase]
+            phase_table.add_row(
+                phase,
+                str(row["fragment_count"]),
+                str(row["resource_count"]),
+                str(row["total_bytes"]),
+                str(row["total_tokens"]),
+                f"{row['budget_used_ratio']:.1%}",
+            )
+        console.print(phase_table)
     else:
         console.print(format_output(data, output_format))