aditya
137d040c4d
CI / benchmark-publish (pull_request) Has been skipped
CI / build (pull_request) Successful in 18s
CI / helm (pull_request) Successful in 22s
CI / quality (pull_request) Successful in 56s
CI / lint (pull_request) Successful in 3m21s
CI / typecheck (pull_request) Successful in 4m0s
CI / security (pull_request) Successful in 4m19s
CI / unit_tests (pull_request) Successful in 9m32s
CI / docker (pull_request) Successful in 1m22s
CI / coverage (pull_request) Successful in 12m27s
CI / e2e_tests (pull_request) Successful in 20m3s
CI / integration_tests (pull_request) Successful in 21m20s
CI / status-check (pull_request) Successful in 1s
CI / benchmark-regression (pull_request) Successful in 54m56s
Add a production skeleton compressor that re-renders inherited fragments to overview depths via the UKO detail-level map chain, fits the result within the configured skeleton budget, and wires the pipeline default to the new compressor. Address prior review feedback by extracting the render visitors into a dedicated module, restoring projected metadata to native runtime types, constraining builtin component resolution with an allowlist, and keeping child-context inheritance compatible with CRP context fragments for the Robot integration path. Reproduced the Forgejo lint job in a clean python:3.13-slim container with the CI commands All checks passed! and 1740 files already formatted; both passed, so the earlier lint failure appears to have been transient runner behavior rather than a source-level defect. ISSUES CLOSED: #919
585 lines
21 KiB
Python
585 lines
21 KiB
Python
"""Step definitions for features/acms_pipeline_phase3.feature.
|
|
|
|
Tests the ACMS pipeline Phase 3 (Context Finalization) components and
|
|
advanced context strategies directly in-memory — no database required.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.application.services.acms_advanced_strategies import (
|
|
ArceStrategy,
|
|
PlanDecisionContextStrategy,
|
|
TemporalArchaeologyStrategy,
|
|
)
|
|
from cleveragents.application.services.acms_phase3 import (
|
|
ProvenancePreambleGenerator,
|
|
RelevanceCoherenceOrderer,
|
|
)
|
|
from cleveragents.application.services.acms_service import ACMSPipeline
|
|
from cleveragents.application.services.acms_skeleton_compressor import (
|
|
DepthReductionCompressor,
|
|
)
|
|
from cleveragents.domain.models.core.context_fragment import (
|
|
ContextBudget,
|
|
ContextFragment,
|
|
FragmentProvenance,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_DEFAULT_PROVENANCE = FragmentProvenance(resource_uri="test://phase3")
|
|
_TEST_PLAN_ID = "01JQTESTPN00000000000000DD"
|
|
|
|
|
|
def _make_phase3_fragment(**kwargs: Any) -> ContextFragment:
|
|
"""Create a ContextFragment with sensible Phase 3 test defaults."""
|
|
kwargs.setdefault("uko_node", "test://default")
|
|
kwargs.setdefault("token_count", 0)
|
|
kwargs.setdefault("provenance", _DEFAULT_PROVENANCE)
|
|
return ContextFragment(**kwargs)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fragment list construction — Given steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the following phase3 fragments:")
|
|
def step_given_phase3_fragments(context: Context) -> None:
|
|
frags: list[ContextFragment] = []
|
|
for row in context.table:
|
|
frags.append(
|
|
_make_phase3_fragment(
|
|
uko_node=row["uko_node"],
|
|
content=row["content"].replace("\\n", "\n"),
|
|
relevance_score=float(row["score"]),
|
|
token_count=int(row["tokens"]),
|
|
detail_depth=int(row["depth"]),
|
|
)
|
|
)
|
|
context.phase3_fragments = frags
|
|
|
|
|
|
@given("the following phase3 fragments with strategy sources:")
|
|
def step_given_phase3_fragments_with_strategy(context: Context) -> None:
|
|
frags: list[ContextFragment] = []
|
|
for row in context.table:
|
|
frags.append(
|
|
_make_phase3_fragment(
|
|
uko_node=row["uko_node"],
|
|
content=row["content"].replace("\\n", "\n"),
|
|
relevance_score=float(row["score"]),
|
|
token_count=int(row["tokens"]),
|
|
detail_depth=int(row["depth"]),
|
|
strategy_source=row["strategy_source"],
|
|
)
|
|
)
|
|
context.phase3_fragments = frags
|
|
|
|
|
|
@given("the following phase3 skeleton fragments:")
|
|
def step_given_phase3_skeleton_fragments(context: Context) -> None:
|
|
frags: list[ContextFragment] = []
|
|
for row in context.table:
|
|
metadata = {}
|
|
if row["domain"].strip():
|
|
metadata["detail_level_domain"] = row["domain"]
|
|
frags.append(
|
|
_make_phase3_fragment(
|
|
uko_node=row["uko_node"],
|
|
content=row["content"].replace("\\n", "\n"),
|
|
relevance_score=float(row["score"]),
|
|
token_count=int(row["tokens"]),
|
|
detail_depth=int(row["depth"]),
|
|
provenance=FragmentProvenance(
|
|
resource_uri=row["uko_node"],
|
|
resource_type=row["resource_type"],
|
|
),
|
|
metadata=metadata,
|
|
)
|
|
)
|
|
context.phase3_skeleton_fragments = tuple(frags)
|
|
|
|
|
|
@given("an empty phase3 fragment list")
|
|
def step_given_empty_phase3(context: Context) -> None:
|
|
context.phase3_fragments = []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Strategy fragment construction — Given steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the following strategy3 fragments:")
|
|
def step_given_strategy3_fragments(context: Context) -> None:
|
|
frags: list[ContextFragment] = []
|
|
for row in context.table:
|
|
frags.append(
|
|
_make_phase3_fragment(
|
|
uko_node=row["uko_node"],
|
|
content=row["content"].replace("\\n", "\n"),
|
|
relevance_score=float(row["score"]),
|
|
token_count=int(row["tokens"]),
|
|
detail_depth=int(row["depth"]),
|
|
)
|
|
)
|
|
context.strategy3_fragments = frags
|
|
|
|
|
|
@given("the following strategy3 fragments with tiers:")
|
|
def step_given_strategy3_fragments_with_tiers(context: Context) -> None:
|
|
frags: list[ContextFragment] = []
|
|
for row in context.table:
|
|
frags.append(
|
|
_make_phase3_fragment(
|
|
uko_node=row["uko_node"],
|
|
content=row["content"].replace("\\n", "\n"),
|
|
relevance_score=float(row["score"]),
|
|
token_count=int(row["tokens"]),
|
|
detail_depth=int(row["depth"]),
|
|
tier=row["tier"],
|
|
)
|
|
)
|
|
context.strategy3_fragments = frags
|
|
|
|
|
|
@given("an empty strategy3 fragment list")
|
|
def step_given_empty_strategy3(context: Context) -> None:
|
|
context.strategy3_fragments = []
|
|
|
|
|
|
@given("a strategy3 budget with max_tokens {max_t:d} and reserved_tokens {res_t:d}")
|
|
def step_given_strategy3_budget(context: Context, max_t: int, res_t: int) -> None:
|
|
context.strategy3_budget = ContextBudget(max_tokens=max_t, reserved_tokens=res_t)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Strategy constructors — Given steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an ArceStrategy with max_iterations {max_iter:d}")
|
|
def step_given_arce_strategy(context: Context, max_iter: int) -> None:
|
|
context.arce_strategy = ArceStrategy(max_iterations=max_iter)
|
|
|
|
|
|
@given("a TemporalArchaeologyStrategy")
|
|
def step_given_temporal_strategy(context: Context) -> None:
|
|
context.temporal_strategy = TemporalArchaeologyStrategy()
|
|
|
|
|
|
@given("a PlanDecisionContextStrategy")
|
|
def step_given_plan_decision_strategy(context: Context) -> None:
|
|
context.plan_decision_strategy = PlanDecisionContextStrategy()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RelevanceCoherenceOrderer — When / Then
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I order the fragments with RelevanceCoherenceOrderer")
|
|
def step_order_fragments(context: Context) -> None:
|
|
orderer = RelevanceCoherenceOrderer()
|
|
context.phase3_result = list(orderer.order(context.phase3_fragments))
|
|
|
|
|
|
@then('the first ordered fragment should have uko_node "{expected}"')
|
|
def step_first_ordered_node(context: Context, expected: str) -> None:
|
|
assert len(context.phase3_result) >= 1
|
|
assert context.phase3_result[0].uko_node == expected, (
|
|
f"Expected {expected}, got {context.phase3_result[0].uko_node}"
|
|
)
|
|
|
|
|
|
@then("fragments from the same group should be adjacent")
|
|
def step_same_group_adjacent(context: Context) -> None:
|
|
# Verify that fragments with the same prefix are adjacent
|
|
seen_prefixes: set[str] = set()
|
|
current_prefix: str | None = None
|
|
for frag in context.phase3_result:
|
|
prefix = _extract_prefix(frag.uko_node)
|
|
if prefix != current_prefix:
|
|
assert prefix not in seen_prefixes, (
|
|
f"Prefix {prefix} appeared non-contiguously"
|
|
)
|
|
if current_prefix is not None:
|
|
seen_prefixes.add(current_prefix)
|
|
current_prefix = prefix
|
|
|
|
|
|
@then("{count:d} fragments should remain after ordering")
|
|
def step_order_count_plural(context: Context, count: int) -> None:
|
|
assert len(context.phase3_result) == count, (
|
|
f"Expected {count}, got {len(context.phase3_result)}"
|
|
)
|
|
|
|
|
|
@then("{count:d} fragment should remain after ordering")
|
|
def step_order_count_singular(context: Context, count: int) -> None:
|
|
assert len(context.phase3_result) == count, (
|
|
f"Expected {count}, got {len(context.phase3_result)}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ProvenancePreambleGenerator — When / Then
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I generate a preamble with ProvenancePreambleGenerator")
|
|
def step_generate_preamble(context: Context) -> None:
|
|
generator = ProvenancePreambleGenerator()
|
|
context.phase3_preamble = generator.generate(context.phase3_fragments)
|
|
|
|
|
|
@then('the preamble should contain "{text}"')
|
|
def step_preamble_contains(context: Context, text: str) -> None:
|
|
assert context.phase3_preamble is not None, "Preamble was None"
|
|
assert text in context.phase3_preamble, (
|
|
f"Expected preamble to contain '{text}', got:\n{context.phase3_preamble}"
|
|
)
|
|
|
|
|
|
@then("the preamble should be None")
|
|
def step_preamble_is_none(context: Context) -> None:
|
|
assert context.phase3_preamble is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DepthReductionCompressor — When / Then
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I compress with DepthReductionCompressor and budget {budget:d}")
|
|
def step_compress_depth_reduction(context: Context, budget: int) -> None:
|
|
compressor = DepthReductionCompressor()
|
|
source = getattr(context, "phase3_skeleton_fragments", ())
|
|
context.phase3_compressed = compressor.compress(source, budget)
|
|
|
|
|
|
@then("all compressed fragments should have detail depth at most {max_depth:d}")
|
|
def step_compressed_depth_max(context: Context, max_depth: int) -> None:
|
|
assert all(f.detail_depth <= max_depth for f in context.phase3_compressed), (
|
|
f"Expected all compressed fragments to have depth <= {max_depth}, "
|
|
f"got {[f.detail_depth for f in context.phase3_compressed]}"
|
|
)
|
|
|
|
|
|
@then("compressed fragments should fit within budget {budget:d}")
|
|
def step_compressed_budget(context: Context, budget: int) -> None:
|
|
used = sum(fragment.token_count for fragment in context.phase3_compressed)
|
|
assert used <= budget, (
|
|
f"Compressed fragments used {used} tokens for budget {budget}"
|
|
)
|
|
|
|
|
|
@then('the compressed fragments should include skeleton level "{level}"')
|
|
def step_compressed_level(context: Context, level: str) -> None:
|
|
levels = {
|
|
fragment.metadata.get("skeleton_target_level")
|
|
for fragment in context.phase3_compressed
|
|
}
|
|
assert level in levels, (
|
|
f"Expected level {level!r}, got {sorted(level for level in levels if level is not None)}"
|
|
)
|
|
|
|
|
|
@then("{count:d} fragments should remain after compression")
|
|
def step_compressed_count(context: Context, count: int) -> None:
|
|
assert len(context.phase3_compressed) == count, (
|
|
f"Expected {count} compressed fragments, got {len(context.phase3_compressed)}"
|
|
)
|
|
|
|
|
|
@then('a compressed fragment should contain "{snippet}"')
|
|
def step_compressed_content_contains(context: Context, snippet: str) -> None:
|
|
contents = [fragment.content for fragment in context.phase3_compressed]
|
|
assert any(snippet in content for content in contents), (
|
|
f"Expected a compressed fragment to contain {snippet!r}, got {contents!r}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ArceStrategy — When / Then
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I assemble with the ArceStrategy")
|
|
def step_assemble_arce(context: Context) -> None:
|
|
context.arce_result = list(
|
|
context.arce_strategy.assemble(
|
|
context.strategy3_fragments,
|
|
context.strategy3_budget,
|
|
)
|
|
)
|
|
|
|
|
|
@when("I check can_handle on ArceStrategy")
|
|
def step_check_arce_can_handle(context: Context) -> None:
|
|
context.arce_confidence = context.arce_strategy.can_handle({})
|
|
|
|
|
|
@then("fragments should be returned by arce strategy")
|
|
def step_arce_has_results(context: Context) -> None:
|
|
assert len(context.arce_result) > 0
|
|
|
|
|
|
@then("the arce result should respect the budget")
|
|
def step_arce_respects_budget(context: Context) -> None:
|
|
total = sum(f.token_count for f in context.arce_result)
|
|
available = context.strategy3_budget.available_tokens
|
|
assert total <= available, f"ARCE used {total} tokens, budget was {available}"
|
|
|
|
|
|
@then("{count:d} fragments should be returned by arce strategy")
|
|
def step_arce_exact_count(context: Context, count: int) -> None:
|
|
assert len(context.arce_result) == count
|
|
|
|
|
|
@then("at most {count:d} fragments should be returned by arce strategy")
|
|
def step_arce_at_most_count(context: Context, count: int) -> None:
|
|
assert len(context.arce_result) <= count, (
|
|
f"Expected at most {count}, got {len(context.arce_result)}"
|
|
)
|
|
|
|
|
|
@then("the arce confidence should be {expected:g}")
|
|
def step_arce_confidence(context: Context, expected: float) -> None:
|
|
assert context.arce_confidence == expected
|
|
|
|
|
|
@then('the ArceStrategy name should be "{expected}"')
|
|
def step_arce_name(context: Context, expected: str) -> None:
|
|
assert context.arce_strategy.name == expected
|
|
|
|
|
|
@then("the ArceStrategy should support semantic search")
|
|
def step_arce_supports_semantic(context: Context) -> None:
|
|
assert context.arce_strategy.capabilities.supports_semantic_search
|
|
|
|
|
|
@then('the ArceStrategy explain should contain "{text}"')
|
|
def step_arce_explain(context: Context, text: str) -> None:
|
|
explanation = context.arce_strategy.explain()
|
|
assert text in explanation, (
|
|
f"Expected explain to contain '{text}', got: {explanation}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TemporalArchaeologyStrategy — When / Then
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I assemble with the TemporalArchaeologyStrategy")
|
|
def step_assemble_temporal(context: Context) -> None:
|
|
context.temporal_result = list(
|
|
context.temporal_strategy.assemble(
|
|
context.strategy3_fragments,
|
|
context.strategy3_budget,
|
|
)
|
|
)
|
|
|
|
|
|
@when("I check can_handle on TemporalArchaeologyStrategy")
|
|
def step_check_temporal_can_handle(context: Context) -> None:
|
|
context.temporal_confidence = context.temporal_strategy.can_handle({})
|
|
|
|
|
|
@then('the first temporal result fragment should have uko_node "{expected}"')
|
|
def step_temporal_first_node(context: Context, expected: str) -> None:
|
|
assert len(context.temporal_result) >= 1
|
|
assert context.temporal_result[0].uko_node == expected, (
|
|
f"Expected {expected}, got {context.temporal_result[0].uko_node}"
|
|
)
|
|
|
|
|
|
@then("{count:d} fragments should be returned by temporal strategy")
|
|
def step_temporal_exact_count(context: Context, count: int) -> None:
|
|
assert len(context.temporal_result) == count
|
|
|
|
|
|
@then("at most {count:d} fragments should be returned by temporal strategy")
|
|
def step_temporal_at_most_count(context: Context, count: int) -> None:
|
|
assert len(context.temporal_result) <= count
|
|
|
|
|
|
@then("the temporal confidence should be {expected:g}")
|
|
def step_temporal_confidence(context: Context, expected: float) -> None:
|
|
assert context.temporal_confidence == expected
|
|
|
|
|
|
@then('the TemporalArchaeologyStrategy name should be "{expected}"')
|
|
def step_temporal_name(context: Context, expected: str) -> None:
|
|
assert context.temporal_strategy.name == expected
|
|
|
|
|
|
@then('the TemporalArchaeologyStrategy explain should contain "{text}"')
|
|
def step_temporal_explain(context: Context, text: str) -> None:
|
|
explanation = context.temporal_strategy.explain()
|
|
assert text in explanation
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PlanDecisionContextStrategy — When / Then
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I assemble with the PlanDecisionContextStrategy")
|
|
def step_assemble_plan_decision(context: Context) -> None:
|
|
context.plan_decision_result = list(
|
|
context.plan_decision_strategy.assemble(
|
|
context.strategy3_fragments,
|
|
context.strategy3_budget,
|
|
)
|
|
)
|
|
|
|
|
|
@when("I check can_handle on PlanDecisionContextStrategy")
|
|
def step_check_plan_decision_can_handle(context: Context) -> None:
|
|
context.plan_decision_confidence = context.plan_decision_strategy.can_handle({})
|
|
|
|
|
|
@then('the first plan_decision result should have uko_node "{expected}"')
|
|
def step_plan_decision_first_node(context: Context, expected: str) -> None:
|
|
assert len(context.plan_decision_result) >= 1
|
|
assert context.plan_decision_result[0].uko_node == expected, (
|
|
f"Expected {expected}, got {context.plan_decision_result[0].uko_node}"
|
|
)
|
|
|
|
|
|
@then("{count:d} fragments should be returned by plan_decision strategy")
|
|
def step_plan_decision_exact_count(context: Context, count: int) -> None:
|
|
assert len(context.plan_decision_result) == count
|
|
|
|
|
|
@then("at most {count:d} fragments should be returned by plan_decision strategy")
|
|
def step_plan_decision_at_most_count(context: Context, count: int) -> None:
|
|
assert len(context.plan_decision_result) <= count
|
|
|
|
|
|
@then("the plan_decision confidence should be {expected:g}")
|
|
def step_plan_decision_confidence(context: Context, expected: float) -> None:
|
|
assert context.plan_decision_confidence == expected
|
|
|
|
|
|
@then('the PlanDecisionContextStrategy name should be "{expected}"')
|
|
def step_plan_decision_name(context: Context, expected: str) -> None:
|
|
assert context.plan_decision_strategy.name == expected
|
|
|
|
|
|
@then('the PlanDecisionContextStrategy explain should contain "{text}"')
|
|
def step_plan_decision_explain(context: Context, text: str) -> None:
|
|
explanation = context.plan_decision_strategy.explain()
|
|
assert text in explanation
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pipeline Registration — Given / When / Then
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an ACMS pipeline for phase3 strategy tests")
|
|
def step_given_phase3_pipeline(context: Context) -> None:
|
|
context.phase3_test_pipeline = ACMSPipeline()
|
|
|
|
|
|
@when("I register ArceStrategy with the pipeline")
|
|
def step_register_arce(context: Context) -> None:
|
|
strategy = ArceStrategy()
|
|
context.phase3_test_pipeline.register_strategy("arce", strategy)
|
|
|
|
|
|
@when("I register all phase3 strategies with the pipeline")
|
|
def step_register_all_phase3(context: Context) -> None:
|
|
context.phase3_test_pipeline.register_strategy("arce", ArceStrategy())
|
|
context.phase3_test_pipeline.register_strategy(
|
|
"temporal-archaeology", TemporalArchaeologyStrategy()
|
|
)
|
|
context.phase3_test_pipeline.register_strategy(
|
|
"plan-decision-context", PlanDecisionContextStrategy()
|
|
)
|
|
|
|
|
|
@then('the phase3 pipeline should have strategy "{name}"')
|
|
def step_phase3_pipeline_has_strategy(context: Context, name: str) -> None:
|
|
assert name in context.phase3_test_pipeline._strategies
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pipeline Integration — Given / When / Then
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the ACMS pipeline with a RelevanceCoherenceOrderer")
|
|
def step_pipeline_with_orderer(context: Context) -> None:
|
|
context.phase3_pipeline = ACMSPipeline(
|
|
orderer=RelevanceCoherenceOrderer(),
|
|
)
|
|
|
|
|
|
@given("the ACMS pipeline with a ProvenancePreambleGenerator")
|
|
def step_pipeline_with_preamble(context: Context) -> None:
|
|
context.phase3_pipeline = ACMSPipeline(
|
|
preamble_generator=ProvenancePreambleGenerator(),
|
|
)
|
|
|
|
|
|
@when("I assemble context through the phase3 pipeline")
|
|
def step_assemble_phase3_pipeline(context: Context) -> None:
|
|
budget = ContextBudget(max_tokens=100000, reserved_tokens=0)
|
|
payload = context.phase3_pipeline.assemble(
|
|
plan_id=_TEST_PLAN_ID,
|
|
fragments=context.phase3_fragments,
|
|
budget=budget,
|
|
)
|
|
context.phase3_payload = payload
|
|
|
|
|
|
@then("the phase3 pipeline output should contain {count:d} fragments")
|
|
def step_phase3_pipeline_count_plural(context: Context, count: int) -> None:
|
|
assert len(context.phase3_payload.fragments) == count, (
|
|
f"Expected {count}, got {len(context.phase3_payload.fragments)}"
|
|
)
|
|
|
|
|
|
@then("the phase3 pipeline output should contain {count:d} fragment")
|
|
def step_phase3_pipeline_count_singular(context: Context, count: int) -> None:
|
|
assert len(context.phase3_payload.fragments) == count
|
|
|
|
|
|
@then("the phase3 pipeline output should have a preamble")
|
|
def step_phase3_pipeline_has_preamble(context: Context) -> None:
|
|
assert context.phase3_payload.preamble is not None, (
|
|
"Expected pipeline output to have a preamble"
|
|
)
|
|
|
|
|
|
@then('the phase3 pipeline preamble should contain "{text}"')
|
|
def step_phase3_pipeline_preamble_contains(context: Context, text: str) -> None:
|
|
preamble = context.phase3_payload.preamble
|
|
assert preamble is not None, "Preamble was None"
|
|
assert text in preamble, f"Expected preamble to contain '{text}', got:\n{preamble}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Internal helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _extract_prefix(uko_node: str) -> str:
|
|
"""Extract grouping prefix for adjacency checks."""
|
|
if "://" in uko_node:
|
|
uko_node = uko_node.split("://", 1)[1]
|
|
segments = [s for s in uko_node.replace("\\", "/").split("/") if s]
|
|
return "/".join(segments[:2]) if len(segments) >= 2 else uko_node
|