cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

fix(lsp): wire lsp_context_enrichment settings into ACMS context assembly pipeline for automatic diagnostic injection #9955

Closed
HAL9000 wants to merge 1 commits from fix/lsp-context-enrichment-acms-wiring into master
pull from: fix/lsp-context-enrichment-acms-wiring
merge into: cleveragents:master
Conversation 39 Commits 1 Files Changed 6
+1140 -1
6 changed files with 1140 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+11
View File
@@ -5,6 +5,17 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
## [Unreleased]
### Added
- **LSP context enrichment wired into ACMS pipeline** (#9191): The
``lsp_context_enrichment`` settings defined in
:class:`~cleveragents.actor.schema.LspContextEnrichment` are now read
and acted upon by the ACMS Context Assembly Pipeline. A new
:class:`~cleveragents.application.services.lsp_context_enricher.LspContextEnricher`
Application-layer service automatically enriches source file fragments
with LSP diagnostic annotations when an actor has
``lsp_context_enrichment.diagnostics: true``.
### Fixed
- **Cross-actor subgraph cycle detection reads actor_ref field** (#1431): Fixed
features/lsp_context_enricher.feature
+171
View File
@@ -0,0 +1,171 @@
@mock_only
Feature: LSP Context Enricher — ACMS hot context diagnostic injection
As an ACMS pipeline
I need to automatically enrich source file fragments with LSP diagnostics
So that actors receive diagnostic annotations without explicitly calling lsp/diagnostics
# ── LspContextEnricher construction ──────────────────────────────────────
Scenario: lce enricher exposes config property
Given lce an LspContextEnricher with diagnostics enabled and server "local/pyright"
Then lce the enricher config diagnostics should be True
Scenario: lce enricher exposes server_names property
Given lce an LspContextEnricher with diagnostics enabled and server "local/pyright"
Then lce the enricher server_names should contain "local/pyright"
Scenario: lce enricher raises ValueError when lsp_runtime is None
When lce I construct an LspContextEnricher with lsp_runtime None
Then lce a ValueError should have been raised
Scenario: lce enricher raises ValueError when config is None
When lce I construct an LspContextEnricher with config None
Then lce a ValueError should have been raised
Scenario: lce enricher raises ValueError when server_names is None
When lce I construct an LspContextEnricher with server_names None
Then lce a ValueError should have been raised
Scenario: lce enricher raises ValueError when max_diagnostics_per_file is negative
When lce I construct an LspContextEnricher with max_diagnostics_per_file -1
Then lce a ValueError should have been raised
# ── Skipping when disabled ────────────────────────────────────────────────
Scenario: lce enricher skips all enrichment when diagnostics and type_annotations are both false
Given lce an LspContextEnricher with diagnostics disabled and type_annotations disabled
And lce a fragment with path "src/main.py" and content "x = 1"
When lce I enrich the fragments
Then lce the enriched fragment content should equal "x = 1"
And lce the LSP runtime should not have been called
Scenario: lce enricher skips when no server names are configured
Given lce an LspContextEnricher with diagnostics enabled and no server names
And lce a fragment with path "src/main.py" and content "x = 1"
When lce I enrich the fragments
Then lce the enriched fragment content should equal "x = 1"
And lce the LSP runtime should not have been called
# ── Diagnostic injection ──────────────────────────────────────────────────
Scenario: lce enricher appends diagnostic block when diagnostics enabled and server returns errors
Given lce an LspContextEnricher with diagnostics enabled and server "local/pyright"
And lce the LSP server returns one ERROR diagnostic for "src/main.py"
And lce a fragment with path "src/main.py" and content "x: int = 'bad'"
When lce I enrich the fragments
Then lce the enriched fragment content should contain "LSP Diagnostics"
And lce the enriched fragment content should contain "ERROR"
And lce the enriched fragment metadata should have "lsp_diagnostics"
Scenario: lce enricher appends no-diagnostics block when server returns empty list
Given lce an LspContextEnricher with diagnostics enabled and server "local/pyright"
And lce the LSP server returns no diagnostics for "src/main.py"
And lce a fragment with path "src/main.py" and content "x = 1"
When lce I enrich the fragments
Then lce the enriched fragment content should contain "LSP Diagnostics"
And lce the enriched fragment content should contain "no diagnostics"
Scenario: lce enricher respects max_diagnostics_per_file cap
Given lce an LspContextEnricher with diagnostics enabled, server "local/pyright", and max_diagnostics 2
And lce the LSP server returns 5 ERROR diagnostics for "src/main.py"
And lce a fragment with path "src/main.py" and content "x = 1"
When lce I enrich the fragments
Then lce the enriched fragment metadata lsp_diagnostics should have at most 2 entries
And lce the enriched fragment content should contain "omitted"
Scenario: lce enricher skips fragments without path metadata
Given lce an LspContextEnricher with diagnostics enabled and server "local/pyright"
And lce a fragment with no path metadata and content "x = 1"
When lce I enrich the fragments
Then lce the enriched fragment content should equal "x = 1"
And lce the LSP runtime should not have been called
Scenario: lce enricher handles LSP server failure gracefully
Given lce an LspContextEnricher with diagnostics enabled and server "local/pyright"
And lce the LSP server raises an error for "src/main.py"
And lce a fragment with path "src/main.py" and content "x = 1"
When lce I enrich the fragments
Then lce the enriched fragment content should equal "x = 1"
Scenario: lce enricher updates token_count after enrichment
Given lce an LspContextEnricher with diagnostics enabled and server "local/pyright"
And lce the LSP server returns one ERROR diagnostic for "src/main.py"
And lce a fragment with path "src/main.py" and content "x = 1"
When lce I enrich the fragments
Then lce the enriched fragment token_count should be greater than the original
# ── Type annotation enrichment ────────────────────────────────────────────
Scenario: lce enricher injects type annotations when type_annotations enabled
Given lce an LspContextEnricher with type_annotations enabled and server "local/pyright"
And lce the LSP server returns hover info for symbol "my_func"
And lce a fragment with path "src/main.py" and content "def my_func(): pass"
When lce I enrich the fragments
Then lce the enriched fragment metadata should have "lsp_type_annotations"
And lce the enriched fragment metadata lsp_type_annotations should contain "my_func"
Scenario: lce enricher skips type annotations when no symbols found
Given lce an LspContextEnricher with type_annotations enabled and server "local/pyright"
And lce a fragment with path "src/main.py" and content "x = 1"
When lce I enrich the fragments
Then lce the enriched fragment metadata should not have "lsp_type_annotations"
Scenario: lce enricher handles hover failure gracefully for type annotations
Given lce an LspContextEnricher with type_annotations enabled and server "local/pyright"
And lce the LSP server raises an error on hover for "src/main.py"
And lce a fragment with path "src/main.py" and content "def my_func(): pass"
When lce I enrich the fragments
Then lce the enriched fragment content should equal "def my_func(): pass"
# ── Multiple fragments ────────────────────────────────────────────────────
Scenario: lce enricher processes multiple fragments independently
Given lce an LspContextEnricher with diagnostics enabled and server "local/pyright"
And lce the LSP server returns one ERROR diagnostic for "src/a.py"
And lce the LSP server returns no diagnostics for "src/b.py"
And lce two fragments with paths "src/a.py" and "src/b.py"
When lce I enrich the fragments
Then lce both enriched fragments should contain "LSP Diagnostics"
# ── ACMSExecutePhaseContextAssembler wiring ───────────────────────────────
Scenario: lce assembler accepts lsp_context_enricher parameter
Given lce an ACMSExecutePhaseContextAssembler with an LspContextEnricher wired in
Then lce the assembler lsp_context_enricher property should not be None
Scenario: lce assembler without enricher has None lsp_context_enricher
Given lce an ACMSExecutePhaseContextAssembler without an LspContextEnricher
Then lce the assembler lsp_context_enricher property should be None
# ── Helper functions ──────────────────────────────────────────────────────
Scenario: lce format_diagnostic formats ERROR severity correctly
When lce I format a diagnostic with severity 1 message "Type error" at line 5 char 3
Then lce the formatted diagnostic should contain "ERROR"
And lce the formatted diagnostic should contain "6:4"
And lce the formatted diagnostic should contain "Type error"
Scenario: lce format_diagnostic formats WARNING severity correctly
When lce I format a diagnostic with severity 2 message "Unused var" at line 0 char 0
Then lce the formatted diagnostic should contain "WARNING"
Scenario: lce format_diagnostic includes source prefix when source present
When lce I format a diagnostic with source "pyright" message "Error" at line 0 char 0
Then lce the formatted diagnostic should contain "[pyright]"
Scenario: lce extract_key_symbols finds function definitions
When lce I extract key symbols from "def foo(): pass\ndef bar(): pass"
Then lce the extracted symbols should contain "foo"
And lce the extracted symbols should contain "bar"
Scenario: lce extract_key_symbols finds class definitions
When lce I extract key symbols from "class MyClass:\n pass"
Then lce the extracted symbols should contain "MyClass"
Scenario: lce extract_key_symbols deduplicates repeated names
When lce I extract key symbols from "def foo(): pass\ndef foo(): pass"
Then lce the extracted symbols should have exactly 1 entry for "foo"
Scenario: lce extract_key_symbols returns empty for plain content
When lce I extract key symbols from "x = 1\ny = 2"
Then lce the extracted symbols should be empty
features/steps/lsp_context_enricher_helpers_steps.py
+233
View File
@@ -0,0 +1,233 @@
"""Shared helpers for lsp_context_enricher BDD step files.
Contains factory functions and unit-test-style When-steps that keep
the main step file under the 500-line limit.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any, cast
from unittest.mock import MagicMock
from behave import when
from behave.runner import Context
from cleveragents.actor.schema import LspContextEnrichment
from cleveragents.lsp.runtime import LspRuntime
if TYPE_CHECKING:
from cleveragents.application.services.lsp_context_enricher import (
LspContextEnricher,
)
# ---------------------------------------------------------------------------
# Factory helpers used by all step modules
# ---------------------------------------------------------------------------
def make_fragment(
path: str | None = "src/main.py",
content: str = "x = 1",
fragment_id: str = "frag-001",
) -> Any:
"""Build a minimal ContextFragment for testing."""
from cleveragents.domain.models.core.context_fragment import (
ContextFragment,
FragmentProvenance,
)
metadata: dict[str, str] = {}
if path is not None:
metadata["path"] = path
return ContextFragment(
uko_node="uko:test",
content=content,
detail_depth=1,
token_count=len(content) // 4 or 1,
relevance_score=0.5,
provenance=FragmentProvenance(
resource_uri="res:test",
location=path or "",
strategy="test",
),
metadata=metadata,
fragment_id=fragment_id,
)
def make_mock_runtime() -> MagicMock:
"""Create a mock LspRuntime."""
runtime = MagicMock(spec=LspRuntime)
runtime.get_diagnostics = MagicMock(return_value=[])
runtime.get_hover = MagicMock(return_value=None)
return runtime
def make_enricher(
runtime: LspRuntime | MagicMock,
diagnostics: bool = True,
type_annotations: bool = False,
max_diagnostics: int = 50,
server_names: list[str] | None = None,
) -> Any:
"""Build an LspContextEnricher with the given settings."""
from cleveragents.application.services.lsp_context_enricher import (
LspContextEnricher as _Lce,
)
config = LspContextEnrichment(
diagnostics=diagnostics,
type_annotations=type_annotations,
max_diagnostics_per_file=max_diagnostics,
)
return _Lce(
lsp_runtime=cast(LspRuntime, runtime),
config=config,
server_names=server_names if server_names is not None else ["local/pyright"],
)
def make_assembler_with_enricher() -> Any:
"""Build a minimal ACMS assembler with a wired enricher."""
from unittest.mock import MagicMock
from cleveragents.application.services.execute_phase_context_assembler import (
ACMSExecutePhaseContextAssembler as _Asm,
)
runtime = make_mock_runtime()
enricher = make_enricher(runtime)
mock_tier = MagicMock()
mock_repo = MagicMock()
return _Asm(
context_tier_service=mock_tier,
project_repository=mock_repo,
lsp_context_enricher=enricher,
)
# ---------------------------------------------------------------------------
# Helper When-steps (unit-test style) — keeps the main step file < 500 lines.
# ---------------------------------------------------------------------------
@when("lce I construct an LspContextEnricher with lsp_runtime None")
def _step_construct_none_runtime(context: Context) -> None:
config = LspContextEnrichment(diagnostics=True)
from cleveragents.application.services.lsp_context_enricher import (
LspContextEnricher as _Lce,
)
try:
_Lce(
lsp_runtime=cast(LspRuntime, None),
config=config,
server_names=["local/pyright"],
)
context.lce_raised_error = None
except ValueError as exc:
context.lce_raised_error = exc
@when("lce I construct an LspContextEnricher with config None")
def _step_construct_none_config(context: Context) -> None:
runtime = make_mock_runtime()
from cleveragents.application.services.lsp_context_enricher import (
LspContextEnricher as _Lce,
)
try:
_Lce(
lsp_runtime=cast(LspRuntime, runtime),
config=cast(LspContextEnrichment, None),
server_names=["local/pyright"],
)
context.lce_raised_error = None
except ValueError as exc:
context.lce_raised_error = exc
@when("lce I construct an LspContextEnricher with server_names None")
def _step_construct_none_server_names(context: Context) -> None:
runtime = make_mock_runtime()
config = LspContextEnrichment(diagnostics=True)
from cleveragents.application.services.lsp_context_enricher import (
LspContextEnricher as _Lce,
)
try:
_Lce(
lsp_runtime=cast(LspRuntime, runtime),
config=config,
server_names=cast(list[str], None),
)
context.lce_raised_error = None
except ValueError as exc:
context.lce_raised_error = exc
@when("lce I construct an LspContextEnricher with max_diagnostics_per_file -1")
def _step_construct_negative_max_diag(context: Context) -> None:
runtime = make_mock_runtime()
config = LspContextEnrichment(diagnostics=True, max_diagnostics_per_file=-1)
from cleveragents.application.services.lsp_context_enricher import (
LspContextEnricher as _Lce,
)
try:
_Lce(
lsp_runtime=cast(LspRuntime, runtime),
config=config,
server_names=["local/pyright"],
)
context.lce_raised_error = None
except ValueError as exc:
context.lce_raised_error = exc
@when(
'lce I format a diagnostic with severity {severity:d} message "{message}" at line {line:d} char {char:d}'
)
def _step_format_diagnostic_severity(
context: Context, severity: int, message: str, line: int, char: int
) -> None:
from cleveragents.application.services.lsp_context_enricher import (
_format_diagnostic as _fd,
)
diag: dict[str, Any] = {
"severity": severity,
"message": message,
"range": {"start": {"line": line, "character": char}},
}
context.lce_formatted = _fd(diag)
@when(
'lce I format a diagnostic with source "{source}" message "{message}" at line {line:d} char {char:d}'
)
def _step_format_diagnostic_source(
context: Context, source: str, message: str, line: int, char: int
) -> None:
from cleveragents.application.services.lsp_context_enricher import (
_format_diagnostic as _fd,
)
diag: dict[str, Any] = {
"severity": 1,
"message": message,
"source": source,
"range": {"start": {"line": line, "character": char}},
}
context.lce_formatted = _fd(diag)
@when('lce I extract key symbols from "{content}"')
def _step_extract_symbols(context: Context, content: str) -> None:
from cleveragents.application.services.lsp_context_enricher import (
_extract_key_symbols as _eks,
)
unescaped = content.replace("\\n", "\n")
context.lce_symbols = _eks(unescaped)
features/steps/lsp_context_enricher_steps.py
+368
View File
@@ -0,0 +1,368 @@
"""Step definitions for lsp_context_enricher.feature.
Tests the LspContextEnricher service and its wiring into
ACMSExecutePhaseContextAssembler using the ``lce`` step prefix to avoid
Behave AmbiguousStep errors.
Helper factory functions (make_fragment, make_mock_runtime, make_enricher)
and unit-test-style When-steps for private helpers are defined in
:py:mod:`lsp_context_enricher_helpers_steps`.
"""
from __future__ import annotations
import json
from typing import Any
from behave import given, then
from behave.runner import Context
from cleveragents.actor.schema import LspContextEnrichment
from cleveragents.application.services.execute_phase_context_assembler import (
ACMSExecutePhaseContextAssembler,
)
from cleveragents.application.services.lsp_context_enricher import (
LspContextEnricher,
)
from cleveragents.domain.models.core.context_fragment import ContextFragment
# Import helper factories for use in Given step bodies.
from lsp_context_enricher_helpers_steps import (
make_assembler_with_enricher,
make_enricher,
make_fragment,
make_mock_runtime,
)
# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------
@given('lce an LspContextEnricher with diagnostics enabled and server "{server}"')
def step_lce_enricher_diag_enabled(context: Context, server: str) -> None:
context.lce_runtime = make_mock_runtime()
context.lce_enricher = make_enricher(
context.lce_runtime, diagnostics=True, server_names=[server]
)
context.lce_fragments: list[ContextFragment] = []
@given(
"lce an LspContextEnricher with diagnostics disabled and type_annotations disabled"
)
def step_lce_enricher_both_disabled(context: Context) -> None:
context.lce_runtime = make_mock_runtime()
context.lce_enricher = make_enricher(
context.lce_runtime,
diagnostics=False,
type_annotations=False,
)
context.lce_fragments = []
@given("lce an LspContextEnricher with diagnostics enabled and no server names")
def step_lce_enricher_no_servers(context: Context) -> None:
context.lce_runtime = make_mock_runtime()
context.lce_enricher = make_enricher(
context.lce_runtime, diagnostics=True, server_names=[]
)
context.lce_fragments = []
@given(
'lce an LspContextEnricher with diagnostics enabled, server "{server}", '
"and max_diagnostics {max_diag:d}"
)
def step_lce_enricher_with_max(context: Context, server: str, max_diag: int) -> None:
context.lce_runtime = make_mock_runtime()
context.lce_enricher = make_enricher(
context.lce_runtime,
diagnostics=True,
max_diagnostics=max_diag,
server_names=[server],
)
context.lce_fragments = []
@given('lce an LspContextEnricher with type_annotations enabled and server "{server}"')
def step_lce_enricher_type_ann_enabled(context: Context, server: str) -> None:
context.lce_runtime = make_mock_runtime()
context.lce_enricher = make_enricher(
context.lce_runtime,
diagnostics=False,
type_annotations=True,
server_names=[server],
)
context.lce_fragments = []
@given('lce the LSP server returns one ERROR diagnostic for "{file_path}"')
def step_lce_server_returns_one_error(context: Context, file_path: str) -> None:
diag: dict[str, Any] = {
"severity": 1,
"message": "Type mismatch",
"range": {"start": {"line": 0, "character": 0}},
"source": "pyright",
}
context.lce_runtime.get_diagnostics.return_value = [diag]
@given('lce the LSP server returns no diagnostics for "{file_path}"')
def step_lce_server_returns_no_diag(context: Context, file_path: str) -> None:
context.lce_runtime.get_diagnostics.return_value = []
@given('lce the LSP server returns {count:d} ERROR diagnostics for "{file_path}"')
def step_lce_server_returns_n_errors(
context: Context, count: int, file_path: str
) -> None:
diags: list[dict[str, Any]] = [
{
"severity": 1,
"message": f"Error {i}",
"range": {"start": {"line": i, "character": 0}},
}
for i in range(count)
]
context.lce_runtime.get_diagnostics.return_value = diags
@given('lce the LSP server raises an error for "{file_path}"')
def step_lce_server_raises_error(context: Context, file_path: str) -> None:
from cleveragents.lsp.errors import LspError
context.lce_runtime.get_diagnostics.side_effect = LspError(
"Server not running", details={}
)
@given('lce the LSP server returns hover info for symbol "{symbol}"')
def step_lce_server_returns_hover(context: Context, symbol: str) -> None:
context.lce_runtime.get_hover.return_value = {
"contents": {
"kind": "markdown",
"value": f"```python\ndef {symbol}() -> None\n```",
}
}
@given('lce the LSP server raises an error on hover for "{file_path}"')
def step_lce_server_raises_hover_error(context: Context, file_path: str) -> None:
from cleveragents.lsp.errors import LspError
context.lce_runtime.get_hover.side_effect = LspError("Hover failed", details={})
@given('lce a fragment with path "{path}" and content "{content}"')
def step_lce_fragment_with_path(context: Context, path: str, content: str) -> None:
context.lce_fragments = [make_fragment(path=path, content=content)]
context.lce_original_content = content
context.lce_original_token_count = context.lce_fragments[0].token_count
@given('lce a fragment with no path metadata and content "{content}"')
def step_lce_fragment_no_path(context: Context, content: str) -> None:
context.lce_fragments = [make_fragment(path=None, content=content)]
context.lce_original_content = content
@given('lce two fragments with paths "{path_a}" and "{path_b}"')
def step_lce_two_fragments(context: Context, path_a: str, path_b: str) -> None:
context.lce_fragments = [
make_fragment(path=path_a, content="x = 1", fragment_id="frag-a"),
make_fragment(path=path_b, content="y = 2", fragment_id="frag-b"),
]
@given("lce an ACMSExecutePhaseContextAssembler with an LspContextEnricher wired in")
def step_lce_assembler_with_enricher(context: Context) -> None:
context.lce_assembler = make_assembler_with_enricher()
@given("lce an ACMSExecutePhaseContextAssembler without an LspContextEnricher")
def step_lce_assembler_without_enricher(context: Context) -> None:
from unittest.mock import MagicMock
mock_tier = MagicMock()
mock_repo = MagicMock()
context.lce_assembler = ACMSExecutePhaseContextAssembler(
context_tier_service=mock_tier,
project_repository=mock_repo,
)
# ---------------------------------------------------------------------------
# When step: enrich (shared across most scenarios)
# ---------------------------------------------------------------------------
@when("lce I enrich the fragments")
def step_lce_enrich(context: Context) -> None:
context.lce_enriched = context.lce_enricher.enrich(context.lce_fragments)
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then("lce a ValueError should have been raised")
def step_lce_value_error_raised(context: Context) -> None:
assert context.lce_raised_error is not None, (
"Expected a ValueError to be raised, but no error was raised"
)
assert isinstance(context.lce_raised_error, ValueError), (
f"Expected ValueError, got {type(context.lce_raised_error)}"
)
@then("lce the enricher config diagnostics should be True")
def step_lce_config_diagnostics_true(context: Context) -> None:
assert context.lce_enricher.config.diagnostics is True, (
f"Expected diagnostics=True, got {context.lce_enricher.config.diagnostics}"
)
@then('lce the enricher server_names should contain "{server}"')
def step_lce_server_names_contain(context: Context, server: str) -> None:
assert server in context.lce_enricher.server_names, (
f"Expected '{server}' in {context.lce_enricher.server_names}"
)
@then('lce the enriched fragment content should equal "{expected}"')
def step_lce_content_equals(context: Context, expected: str) -> None:
assert len(context.lce_enriched) == 1, (
f"Expected 1 fragment, got {len(context.lce_enriched)}"
)
actual = context.lce_enriched[0].content
assert actual == expected, f"Expected content {expected!r}, got {actual!r}"
@then('lce the enriched fragment content should contain "{substring}"')
def step_lce_content_contains(context: Context, substring: str) -> None:
assert len(context.lce_enriched) >= 1, "No enriched fragments"
actual = context.lce_enriched[0].content
assert substring in actual, f"Expected {substring!r} in content, got:\n{actual}"
@then("lce the LSP runtime should not have been called")
def step_lce_runtime_not_called(context: Context) -> None:
context.lce_runtime.get_diagnostics.assert_not_called()
context.lce_runtime.get_hover.assert_not_called()
@then('lce the enriched fragment metadata should have "{key}"')
def step_lce_metadata_has_key(context: Context, key: str) -> None:
assert len(context.lce_enriched) >= 1, "No enriched fragments"
metadata = context.lce_enriched[0].metadata
assert isinstance(metadata, dict), f"Expected dict metadata, got {type(metadata)}"
assert key in metadata, (
f"Expected key '{key}' in metadata, got keys: {list(metadata.keys())}"
)
@then('lce the enriched fragment metadata should not have "{key}"')
def step_lce_metadata_not_has_key(context: Context, key: str) -> None:
assert len(context.lce_enriched) >= 1, "No enriched fragments"
metadata = context.lce_enriched[0].metadata
assert isinstance(metadata, dict), f"Expected dict metadata, got {type(metadata)}"
assert key not in metadata, (
f"Expected key '{key}' NOT in metadata, but it was present"
)
@then(
"lce the enriched fragment metadata lsp_diagnostics should have "
"at most {max_count:d} entries"
)
def step_lce_metadata_diag_capped(context: Context, max_count: int) -> None:
assert len(context.lce_enriched) >= 1, "No enriched fragments"
metadata = context.lce_enriched[0].metadata
assert isinstance(metadata, dict), "Expected dict metadata"
diags_raw = metadata.get("lsp_diagnostics", "[]")
diags = json.loads(diags_raw) if isinstance(diags_raw, str) else []
assert len(diags) <= max_count, (
f"Expected at most {max_count} diagnostics, got {len(diags)}"
)
@then("lce the enriched fragment token_count should be greater than the original")
def step_lce_token_count_increased(context: Context) -> None:
assert len(context.lce_enriched) >= 1, "No enriched fragments"
new_count = context.lce_enriched[0].token_count
original = context.lce_original_token_count
assert new_count > original, f"Expected token_count > {original}, got {new_count}"
@then(
'lce the enriched fragment metadata lsp_type_annotations should contain "{symbol}"'
)
def step_lce_type_ann_contains_symbol(context: Context, symbol: str) -> None:
assert len(context.lce_enriched) >= 1, "No enriched fragments"
metadata = context.lce_enriched[0].metadata
assert isinstance(metadata, dict), "Expected dict metadata"
ann_raw = metadata.get("lsp_type_annotations", "{}")
annotations = json.loads(ann_raw) if isinstance(ann_raw, str) else {}
assert symbol in annotations, (
f"Expected symbol '{symbol}' in type annotations, "
f"got: {list(annotations.keys())}"
)
@then('lce both enriched fragments should contain "LSP Diagnostics"')
def step_lce_both_contain_diag(context: Context) -> None:
assert len(context.lce_enriched) == 2, (
f"Expected 2 enriched fragments, got {len(context.lce_enriched)}"
)
for frag in context.lce_enriched:
assert "LSP Diagnostics" in frag.content, (
f"Expected 'LSP Diagnostics' in fragment content:\n{frag.content}"
)
@then("lce the assembler lsp_context_enricher property should not be None")
def step_lce_assembler_enricher_not_none(context: Context) -> None:
assert context.lce_assembler.lsp_context_enricher is not None, (
"Expected lsp_context_enricher to be set, got None"
)
@then("lce the assembler lsp_context_enricher property should be None")
def step_lce_assembler_enricher_is_none(context: Context) -> None:
assert context.lce_assembler.lsp_context_enricher is None, (
"Expected lsp_context_enricher to be None, "
f"got {context.lce_assembler.lsp_context_enricher}"
)
@then('lce the formatted diagnostic should contain "{substring}"')
def step_lce_formatted_contains(context: Context, substring: str) -> None:
assert substring in context.lce_formatted, (
f"Expected '{substring}' in formatted diagnostic: {context.lce_formatted!r}"
)
@then('lce the extracted symbols should contain "{symbol}"')
def step_lce_symbols_contain(context: Context, symbol: str) -> None:
assert symbol in context.lce_symbols, (
f"Expected '{symbol}' in symbols: {context.lce_symbols}"
)
@then("lce the extracted symbols should be empty")
def step_lce_symbols_empty(context: Context) -> None:
assert context.lce_symbols == [], (
f"Expected empty symbols, got: {context.lce_symbols}"
)
@then('lce the extracted symbols should have exactly 1 entry for "{symbol}"')
def step_lce_symbols_deduped(context: Context, symbol: str) -> None:
count = context.lce_symbols.count(symbol)
assert count == 1, (
f"Expected exactly 1 occurrence of '{symbol}', "
f"got {count} in {context.lce_symbols}"
)
src/cleveragents/application/services/execute_phase_context_assembler.py
+17 -1
View File
@@ -4,7 +4,7 @@ from __future__ import annotations
import fnmatch
from pathlib import PurePath
from typing import Any, Protocol
from typing import TYPE_CHECKING, Any, Protocol
import structlog
@@ -25,6 +25,11 @@ from cleveragents.infrastructure.database.repositories import (
NamespacedProjectRepository,
)
if TYPE_CHECKING:
from cleveragents.application.services.lsp_context_enricher import (
LspContextEnricher,
)
logger = structlog.get_logger(__name__)
@@ -45,11 +50,13 @@ class ACMSExecutePhaseContextAssembler(ExecutePhaseContextAssembler):
project_repository: NamespacedProjectRepository,
acms_pipeline: ACMSPipeline | None = None,
hot_max_tokens: int = 4096,
lsp_context_enricher: LspContextEnricher | None = None,
) -> None:
self._tier = context_tier_service
self._project_repository = project_repository
self._pipeline = acms_pipeline or ACMSPipeline()
self._hot_max_tokens = hot_max_tokens
self._lsp_enricher = lsp_context_enricher
self._logger = logger.bind(component="execute_phase_context_assembler")
@property
@@ -57,6 +64,11 @@ class ACMSExecutePhaseContextAssembler(ExecutePhaseContextAssembler):
"""Public accessor for the underlying ContextTierService."""
return self._tier
@property
def lsp_context_enricher(self) -> LspContextEnricher | None:
"""The optional LSP context enricher wired into this assembler."""
return self._lsp_enricher
def _resolve_execute_view(self, project_name: str) -> Any:
"""Resolve the effective execute-phase view for *project_name*."""
try:
@@ -207,6 +219,10 @@ class ACMSExecutePhaseContextAssembler(ExecutePhaseContextAssembler):
)
return None
# Apply LSP context enrichment post-filter, pre-pipeline.
if self._lsp_enricher is not None:
filtered = self._lsp_enricher.enrich(filtered)
budget = CoreContextBudget(max_tokens=self._hot_max_tokens, reserved_tokens=0)
request = ContextRequest(
query=(
src/cleveragents/application/services/lsp_context_enricher.py
+340
View File
@@ -0,0 +1,340 @@
"""LSP Context Enricher — automatic diagnostic injection into ACMS context.
The :class:`LspContextEnricher` reads ``lsp_context_enrichment`` settings from
an actor's configuration and enriches ACMS :class:`ContextFragment` objects
with LSP diagnostics and type annotations before they enter the ACMS pipeline.
This service is wired into :class:`ACMSExecutePhaseContextAssembler` as a
post-filter, pre-pipeline enrichment step so that actors with LSP bindings
automatically receive diagnostic annotations without explicitly calling
``lsp/diagnostics``.
Based on ``docs/specification.md`` LSP Context Enrichment (issue #9191).
"""
from __future__ import annotations
import json
import re
from typing import TYPE_CHECKING, Any
import structlog
from cleveragents.actor.schema import LspContextEnrichment
from cleveragents.domain.models.core.context_fragment import ContextFragment
if TYPE_CHECKING:
from cleveragents.lsp.runtime import LspRuntime
logger = structlog.get_logger(__name__)
# Maximum number of symbols to fetch hover info for per fragment.
_MAX_HOVER_SYMBOLS: int = 10
# Compiled regex for extracting function/class definitions.
_SYMBOL_RE: re.Pattern[str] = re.compile(r"^(?:def|class)\s+(\w+)", re.MULTILINE)
_SEVERITY_MAP: dict[int, str] = {
1: "ERROR",
2: "WARNING",
3: "INFORMATION",
4: "HINT",
}
def _format_diagnostic(diag: dict[str, Any]) -> str:
"""Format a single LSP diagnostic dict into a human-readable string.
Args:
diag: LSP Diagnostic dict with ``severity``, ``message``, ``range``,
and optional ``source`` fields.
Returns:
A formatted string like ``"ERROR 1:1 [pyright] Type mismatch"``.
"""
severity_raw = diag.get("severity", 1)
try:
severity = _SEVERITY_MAP.get(int(severity_raw), "ERROR")
except (ValueError, TypeError):
severity = "ERROR"
range_info = diag.get("range", {})
start = range_info.get("start", {})
line = start.get("line", 0) + 1 # Convert 0-based to 1-based
char = start.get("character", 0) + 1 # Convert 0-based to 1-based
message = diag.get("message", "")
source = diag.get("source", "")
source_prefix = f"[{source}] " if source else ""
return f"{severity} {line}:{char} {source_prefix}{message}"
def _format_diagnostics_block(
diagnostics: list[dict[str, Any]],
max_count: int,
) -> str:
"""Format a list of diagnostics into a block string.
Args:
diagnostics: List of LSP Diagnostic dicts.
max_count: Maximum number of diagnostics to include.
Returns:
A formatted block string with diagnostic lines.
"""
if not diagnostics:
return "# LSP Diagnostics\n# (no diagnostics)\n"
capped = diagnostics[:max_count]
lines = ["# LSP Diagnostics"]
for diag in capped:
lines.append(f"# {_format_diagnostic(diag)}")
omitted = len(diagnostics) - len(capped)
if omitted > 0:
lines.append(f"# ... {omitted} more diagnostic(s) omitted")
return "\n".join(lines) + "\n"
def _extract_key_symbols(content: str) -> list[str]:
"""Extract unique function and class names from Python source content.
Args:
content: Source code text.
Returns:
Deduplicated list of symbol names in order of first appearance.
"""
seen: set[str] = set()
result: list[str] = []
for match in _SYMBOL_RE.finditer(content):
name = match.group(1)
if name not in seen:
seen.add(name)
result.append(name)
return result
class LspContextEnricher:
"""Enriches ACMS context fragments with LSP diagnostics and type info.
Reads ``lsp_context_enrichment`` settings from the actor config and, for
each source file fragment with a ``path`` in its metadata, calls
:meth:`LspRuntime.get_diagnostics` and appends structured diagnostic
annotations to the fragment content.
Args:
lsp_runtime: The LSP runtime to query for diagnostics and hover info.
config: The ``LspContextEnrichment`` settings from the actor config.
server_names: List of LSP server names to query.
Raises:
ValueError: If any required argument is ``None`` or
``max_diagnostics_per_file`` is negative.
"""
def __init__(
self,
lsp_runtime: LspRuntime,
config: LspContextEnrichment,
server_names: list[str],
) -> None:
if lsp_runtime is None:
raise ValueError("lsp_runtime must not be None")
if config is None:
raise ValueError("config must not be None")
if server_names is None:
raise ValueError("server_names must not be None")
if config.max_diagnostics_per_file < 0:
raise ValueError(
f"max_diagnostics_per_file must be >= 0, "
f"got {config.max_diagnostics_per_file}"
)
self._runtime = lsp_runtime
self._config = config
self._server_names = list(server_names)
self._logger = logger.bind(component="lsp_context_enricher")
@property
def config(self) -> LspContextEnrichment:
"""The ``LspContextEnrichment`` configuration."""
return self._config
@property
def server_names(self) -> list[str]:
"""Defensive copy of the configured LSP server names."""
return list(self._server_names)
def enrich(self, fragments: list[ContextFragment]) -> list[ContextFragment]:
"""Enrich a list of context fragments with LSP diagnostics.
For each fragment that has a ``path`` in its metadata, queries the
configured LSP servers for diagnostics and appends a structured
diagnostic block to the fragment content. Fragments without a path
or when enrichment is disabled are returned unchanged.
Args:
fragments: List of context fragments to enrich.
Returns:
New list of context fragments with enriched content and metadata.
"""
if not self._config.diagnostics and not self._config.type_annotations:
return fragments
if not self._server_names:
return fragments
enriched: list[ContextFragment] = []
for fragment in fragments:
try:
enriched.append(self._enrich_fragment(fragment))
except Exception:
self._logger.warning(
"lsp_enrichment_failed",
fragment_id=fragment.fragment_id,
exc_info=True,
)
enriched.append(fragment)
return enriched
def _enrich_fragment(self, fragment: ContextFragment) -> ContextFragment:
"""Enrich a single fragment with LSP data.
Args:
fragment: The context fragment to enrich.
Returns:
Enriched fragment or the original if no path is present.
"""
raw_meta = fragment.metadata
if not isinstance(raw_meta, dict):
return fragment
path_value = raw_meta.get("path")
if not isinstance(path_value, str) or not path_value:
return fragment
metadata: dict[str, str] = dict(raw_meta)
new_content = fragment.content
if self._config.diagnostics:
diagnostics = self._collect_diagnostics(path_value)
if diagnostics is not None:
diag_block = _format_diagnostics_block(
diagnostics, self._config.max_diagnostics_per_file
)
new_content = new_content + "\n" + diag_block
metadata["lsp_diagnostics"] = json.dumps(
diagnostics[: self._config.max_diagnostics_per_file]
)
if self._config.type_annotations:
annotations = self._collect_type_annotations(path_value, fragment.content)
if annotations:
metadata["lsp_type_annotations"] = json.dumps(annotations)
new_metadata: dict[str, str] = dict(metadata)
new_token_count = len(new_content) // 4 or 1
return fragment.model_copy(
update={
"content": new_content,
"metadata": new_metadata,
"token_count": new_token_count,
}
)
def _collect_diagnostics(self, file_path: str) -> list[dict[str, Any]] | None:
"""Collect diagnostics from all configured LSP servers.
Args:
file_path: Path to the file to diagnose.
Returns:
Combined list of diagnostics from all servers, or ``None`` if all
servers failed.
"""
all_diagnostics: list[dict[str, Any]] = []
any_success = False
for server_name in self._server_names:
try:
diags = self._runtime.get_diagnostics(server_name, file_path)
all_diagnostics.extend(diags)
any_success = True
except Exception:
self._logger.warning(
"lsp_diagnostics_server_failed",
server=server_name,
file=file_path,
exc_info=True,
)
if not any_success:
return None
return all_diagnostics
def _collect_type_annotations(self, file_path: str, content: str) -> dict[str, str]:
"""Collect type annotation hover info for key symbols.
Args:
file_path: Path to the file.
content: Source content to extract symbols from.
Returns:
Dict mapping symbol names to their hover info strings.
"""
symbols = _extract_key_symbols(content)
if not symbols:
return {}
annotations: dict[str, str] = {}
lines = content.splitlines()
for symbol in symbols[:_MAX_HOVER_SYMBOLS]:
# Find the line and column of the symbol definition.
symbol_line: int | None = None
symbol_col: int | None = None
for line_idx, line in enumerate(lines):
match = re.search(
r"(?:def|class)\s+(" + re.escape(symbol) + r")\b", line
)
if match:
symbol_line = line_idx + 1
# Use match.start(1) to get the capture group position
# (the symbol name), not the keyword position.
symbol_col = match.start(1) + 1 # 1-based column
break
if symbol_line is None or symbol_col is None:
continue
for server_name in self._server_names:
try:
hover = self._runtime.get_hover(
server_name, file_path, symbol_line, symbol_col
)
if hover:
contents = hover.get("contents", {})
if isinstance(contents, dict):
hover_text = contents.get("value", "")
elif isinstance(contents, str):
hover_text = contents
else:
hover_text = str(contents)
if hover_text:
annotations[symbol] = hover_text
break
except Exception:
self._logger.warning(
"lsp_hover_server_failed",
server=server_name,
symbol=symbol,
file=file_path,
exc_info=True,
)
return annotations
__all__ = ["LspContextEnricher"]