Merge pull request 'feat(acms): implement builtin/context skill for CRP' (#1149) from feature/m5-crp-context-skill into master

Reviewed-on: #1149
Reviewed-by: Jeffrey Phillips Freeman <jeffrey.freeman@cleverthis.com>

features/crp_models.feature

@@ -178,19 +178,60 @@ Feature: CRP (Context Request Protocol) Domain Models
   Scenario: AssembledContext rejects negative total_tokens
     Then creating an AssembledContext with total_tokens -1 should raise ValueError
 
-  # ---- Context Skill Tool Stubs ----
+  # ---- Context Skill Tools (Wired to ACMS Pipeline) ----
 
-  Scenario: request_context tool raises NotImplementedError
-    When I call the request_context tool handler
-    Then it should raise NotImplementedError with message containing "ACMS pipeline"
+  Scenario: request_context returns assembled context with fragments
+    Given a context tier service with sample fragments
+    When I call the request_context tool with purpose "test context"
+    Then the request_context result should contain "fragments" key
+    And the request_context result should contain "total_tokens" key
 
-  Scenario: query_history tool raises NotImplementedError
-    When I call the query_history tool handler
-    Then it should raise NotImplementedError with message containing "ACMS pipeline"
+  Scenario: request_context returns empty result with no stored fragments
+    Given an empty context tier service
+    When I call the request_context tool with purpose "test context"
+    Then the request_context result "fragments" should be an empty list
+    And the request_context result "total_tokens" should be 0
 
-  Scenario: get_context_budget tool raises NotImplementedError
-    When I call the get_context_budget tool handler
-    Then it should raise NotImplementedError with message containing "ACMS pipeline"
+  Scenario: request_context filters fragments by query
+    Given a context tier service with sample fragments
+    When I call the request_context tool filtering by query "auth"
+    Then the request_context result "fragments" should only contain matching content
+
+  Scenario: request_context filters fragments by focus targets
+    Given a context tier service with sample fragments
+    When I call the request_context tool with focus on "auth"
+    Then the request_context result "fragments" should only contain matching content
+
+  Scenario: query_history returns entries for stored fragments
+    Given a context tier service with sample fragments
+    When I call the query_history tool with query "auth"
+    Then the query_history result should contain "entries" key
+    And the query_history result "entries" should not be empty
+
+  Scenario: query_history returns empty entries when no match
+    Given a context tier service with sample fragments
+    When I call the query_history tool with query "nonexistent_xyz_content"
+    Then the query_history result "entries" should be an empty list
+
+  Scenario: get_context_budget returns budget state
+    Given an empty context tier service
+    When I call the get_context_budget tool
+    Then the budget result should contain "max_tokens" key
+    And the budget result should contain "reserved_tokens" key
+    And the budget result should contain "available_tokens" key
+    And the budget result should contain "used_tokens" key
+    And the budget result "used_tokens" should be 0
+
+  Scenario: get_context_budget reports used tokens from hot tier
+    Given a context tier service with sample fragments
+    When I call the get_context_budget tool
+    Then the budget result "used_tokens" should be greater than 0
+
+  Scenario: builtin/context SkillDefinition can be built
+    When I build the builtin/context SkillDefinition
+    Then the skill definition name should be "builtin/context"
+    And the skill definition should have 3 resolved tools
+    And the skill definition should be read-only
 
   Scenario: Context skill tools are registered in the tool registry
     Given a CRP tool registry with context tools registered

features/steps/crp_models_steps.py

@@ -20,6 +20,7 @@ from cleveragents.skills.builtins.context_ops import (
     _handle_get_context_budget,
     _handle_query_history,
     _handle_request_context,
+    build_context_skill_definition,
     register_skill_context_tools,
 )
 from cleveragents.tool.registry import ToolRegistry
@@ -557,42 +558,207 @@ def step_then_assembled_bad_total(context: Any, total: int) -> None:
 
 
 # ---------------------------------------------------------------------------
-# Context skill tool stubs
+# Context skill tools (wired to ACMS pipeline)
 # ---------------------------------------------------------------------------
 
 
-@when("I call the request_context tool handler")
-def step_when_call_request_context(context: Any) -> None:
-    context.tool_error = None
-    try:
-        _handle_request_context({"purpose": "test"})
-    except NotImplementedError as exc:
-        context.tool_error = exc
+def _setup_di_overrides(tier_service: Any) -> None:
+    """Override DI providers for test isolation."""
+    from cleveragents.application.container import get_container, reset_container
+    from cleveragents.application.services.acms_service import ACMSPipeline
+
+    reset_container()
+    container = get_container()
+    from dependency_injector import providers
+
+    container.context_tier_service.override(providers.Object(tier_service))
+    container.acms_pipeline.override(providers.Object(ACMSPipeline()))
 
 
-@when("I call the query_history tool handler")
-def step_when_call_query_history(context: Any) -> None:
-    context.tool_error = None
-    try:
-        _handle_query_history({"query": "test"})
-    except NotImplementedError as exc:
-        context.tool_error = exc
+def _teardown_di_overrides() -> None:
+    """Reset DI container after test."""
+    from cleveragents.application.container import reset_container
+
+    reset_container()
 
 
-@when("I call the get_context_budget tool handler")
+def _make_tier_service_with_fragments() -> Any:
+    """Create a ContextTierService pre-loaded with sample fragments."""
+    from cleveragents.application.services.context_tiers import ContextTierService
+    from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment
+
+    svc = ContextTierService()
+    svc.store(
+        TieredFragment(
+            fragment_id="frag-auth-001",
+            content="class AuthManager: handles authentication",
+            tier=ContextTier.HOT,
+            resource_id="uko-py:module/auth",
+            token_count=50,
+        )
+    )
+    svc.store(
+        TieredFragment(
+            fragment_id="frag-db-002",
+            content="class DatabaseService: handles persistence",
+            tier=ContextTier.WARM,
+            resource_id="uko-py:module/database",
+            token_count=45,
+        )
+    )
+    svc.store(
+        TieredFragment(
+            fragment_id="frag-api-003",
+            content="class APIRouter: routes HTTP requests",
+            tier=ContextTier.COLD,
+            resource_id="uko-py:module/api",
+            token_count=40,
+        )
+    )
+    return svc
+
+
+@given("a context tier service with sample fragments")
+def step_given_tier_service_with_fragments(context: Any) -> None:
+    tier_svc = _make_tier_service_with_fragments()
+    _setup_di_overrides(tier_svc)
+    context.add_cleanup(_teardown_di_overrides)
+
+
+@given("an empty context tier service")
+def step_given_empty_tier_service(context: Any) -> None:
+    from cleveragents.application.services.context_tiers import ContextTierService
+
+    tier_svc = ContextTierService()
+    _setup_di_overrides(tier_svc)
+    context.add_cleanup(_teardown_di_overrides)
+
+
+@when('I call the request_context tool with purpose "{purpose}"')
+def step_when_call_request_context(context: Any, purpose: str) -> None:
+    context.request_context_result = _handle_request_context({"purpose": purpose})
+
+
+@when('I call the request_context tool filtering by query "{query}"')
+def step_when_call_request_context_with_query(context: Any, query: str) -> None:
+    context.request_context_result = _handle_request_context(
+        {"purpose": "filtered query", "query": query}
+    )
+
+
+@when('I call the request_context tool with focus on "{focus}"')
+def step_when_call_request_context_with_focus(context: Any, focus: str) -> None:
+    context.request_context_result = _handle_request_context(
+        {"purpose": "focused query", "focus": [focus]}
+    )
+
+
+@then('the request_context result should contain "{key}" key')
+def step_then_request_context_has_key(context: Any, key: str) -> None:
+    assert key in context.request_context_result, (
+        f"Key '{key}' not in result: {context.request_context_result}"
+    )
+
+
+@then('the request_context result "{key}" should be an empty list')
+def step_then_request_context_empty_list(context: Any, key: str) -> None:
+    assert context.request_context_result[key] == [], (
+        f"Expected empty list, got {context.request_context_result[key]}"
+    )
+
+
+@then('the request_context result "{key}" should be {value:d}')
+def step_then_request_context_int_value(context: Any, key: str, value: int) -> None:
+    assert context.request_context_result[key] == value, (
+        f"Expected {value}, got {context.request_context_result[key]}"
+    )
+
+
+@then('the request_context result "fragments" should only contain matching content')
+def step_then_request_context_filtered(context: Any) -> None:
+    fragments = context.request_context_result["fragments"]
+    for frag in fragments:
+        content_lower = frag["content"].lower()
+        node_lower = frag.get("uko_node", "").lower()
+        assert "auth" in content_lower or "auth" in node_lower, (
+            f"Fragment does not match query 'auth': {frag}"
+        )
+
+
+@when('I call the query_history tool with query "{query}"')
+def step_when_call_query_history(context: Any, query: str) -> None:
+    context.query_history_result = _handle_query_history({"query": query})
+
+
+@then('the query_history result should contain "{key}" key')
+def step_then_query_history_has_key(context: Any, key: str) -> None:
+    assert key in context.query_history_result, (
+        f"Key '{key}' not in result: {context.query_history_result}"
+    )
+
+
+@then('the query_history result "{key}" should not be empty')
+def step_then_query_history_not_empty(context: Any, key: str) -> None:
+    assert len(context.query_history_result[key]) > 0, (
+        f"Expected non-empty list for '{key}'"
+    )
+
+
+@then('the query_history result "{key}" should be an empty list')
+def step_then_query_history_empty_list(context: Any, key: str) -> None:
+    assert context.query_history_result[key] == [], (
+        f"Expected empty list, got {context.query_history_result[key]}"
+    )
+
+
+@when("I call the get_context_budget tool")
 def step_when_call_get_budget(context: Any) -> None:
-    context.tool_error = None
-    try:
-        _handle_get_context_budget({})
-    except NotImplementedError as exc:
-        context.tool_error = exc
+    context.budget_result = _handle_get_context_budget({})
 
 
-@then('it should raise NotImplementedError with message containing "{text}"')
-def step_then_not_impl_error(context: Any, text: str) -> None:
-    assert context.tool_error is not None, "Expected NotImplementedError"
-    assert text in str(context.tool_error), (
-        f"Expected '{text}' in '{context.tool_error}'"
+@then('the budget result should contain "{key}" key')
+def step_then_budget_has_key(context: Any, key: str) -> None:
+    assert key in context.budget_result, (
+        f"Key '{key}' not in result: {context.budget_result}"
+    )
+
+
+@then('the budget result "{key}" should be {value:d}')
+def step_then_budget_int_value(context: Any, key: str, value: int) -> None:
+    assert context.budget_result[key] == value, (
+        f"Expected {value}, got {context.budget_result[key]}"
+    )
+
+
+@then('the budget result "{key}" should be greater than {value:d}')
+def step_then_budget_greater_than(context: Any, key: str, value: int) -> None:
+    assert context.budget_result[key] > value, (
+        f"Expected > {value}, got {context.budget_result[key]}"
+    )
+
+
+@when("I build the builtin/context SkillDefinition")
+def step_when_build_skill_definition(context: Any) -> None:
+    context.context_skill_def = build_context_skill_definition()
+
+
+@then('the skill definition name should be "{name}"')
+def step_then_skill_def_name(context: Any, name: str) -> None:
+    assert context.context_skill_def.skill.name == name, (
+        f"Expected name '{name}', got '{context.context_skill_def.skill.name}'"
+    )
+
+
+@then("the skill definition should have {count:d} resolved tools")
+def step_then_skill_def_tool_count(context: Any, count: int) -> None:
+    actual = len(context.context_skill_def.resolved_tools)
+    assert actual == count, f"Expected {count} tools, got {actual}"
+
+
+@then("the skill definition should be read-only")
+def step_then_skill_def_read_only(context: Any) -> None:
+    assert context.context_skill_def.metadata.read_only is True, (
+        "Expected skill definition to be read-only"
     )
 
 

src/cleveragents/application/container.py

@@ -17,6 +17,7 @@ import structlog
 from dependency_injector import containers, providers
 
 from cleveragents.actor.registry import ActorRegistry
+from cleveragents.application.services.acms_service import ACMSPipeline
 from cleveragents.application.services.actor_service import ActorService
 from cleveragents.application.services.audit_event_subscriber import (
     AuditEventSubscriber,
@@ -747,6 +748,14 @@ class Container(containers.DeclarativeContainer):
         vector_backend=index_vector_backend,
     )
 
+    # ACMS Pipeline — context assembly pipeline (Forgejo #873).
+    # Singleton ensures consistent strategy registry across callers.
+    acms_pipeline = providers.Singleton(
+        ACMSPipeline,
+        settings=settings,
+        unit_of_work=unit_of_work,
+    )
+
     # ACMS ResourceFileWatcher — watches resource files for changes (#578).
     # Reads ``index.auto-reindex`` from ConfigService at construction.
     resource_file_watcher = providers.Singleton(

src/cleveragents/application/services/context_tiers.py

@@ -213,6 +213,21 @@ class ContextTierService(TierRuntimeMixin, ScopedTierMixin):
 
         return None
 
+    def get_all_fragments(self) -> list[TieredFragment]:
+        """Return all fragments across hot/warm/cold tiers.
+
+        This exposes a stable public read API for callers that need a
+        snapshot across all tiers without reaching into private stores.
+        """
+        fragments: list[TieredFragment] = []
+        for store in (self._hot, self._warm, self._cold):
+            fragments.extend(store.values())
+        return fragments
+
+    def get_hot_fragments(self) -> list[TieredFragment]:
+        """Return all fragments currently in the hot tier."""
+        return list(self._hot.values())
+
     # ------------------------------------------------------------------
     # Actor views
     # ------------------------------------------------------------------

src/cleveragents/skills/builtins/__init__.py

@@ -24,6 +24,14 @@ register_skill_file_tools(registry)
 register_skill_search_tools(registry)
 register_skill_context_tools(registry)
 ```
+
+## SkillDefinition Factory
+
+```python
+from cleveragents.skills.builtins.context_ops import build_context_skill_definition
+
+skill_def = build_context_skill_definition()
+```
 """
 
 from __future__ import annotations

src/cleveragents/skills/builtins/context_ops.py

@@ -3,25 +3,36 @@
 Provides the ``builtin/context`` skill tools that actors use to issue
 context requests during reasoning via the Context Request Protocol:
 
-- **request_context**: Request specific context to be added to the
-  conversation.  Accepts query, focus, breadth, depth, and purpose
-  parameters.  Stubbed to raise ``NotImplementedError`` until the
-  ACMS pipeline is wired.
-- **query_history**: Query historical context about past decisions
-  and changes.  Stubbed to raise ``NotImplementedError``.
-- **get_context_budget**: Check remaining context token budget.
-  Stubbed to raise ``NotImplementedError``.
+- **request_context**: Request specific context via the ACMS pipeline.
+  Accepts query, focus, breadth, depth, purpose, and an optional
+  plan_id.  Delegates to ``ACMSPipeline.assemble()`` with fragments
+  sourced from the ``ContextTierService``.
+- **query_history**: Query historical context fragments stored across
+  hot/warm/cold tiers.  Filters by keyword match on fragment content.
+- **get_context_budget**: Return the current context token budget
+  including max, reserved, available, and used token counts.
 
 ## Registration
 
 ```python
-from cleveragents.skills.builtins.context_ops import register_skill_context_tools
+from cleveragents.skills.builtins.context_ops import (
+    register_skill_context_tools,
+)
 from cleveragents.tool.registry import ToolRegistry
 
 registry = ToolRegistry()
 register_skill_context_tools(registry)
 ```
 
+## SkillDefinition
+
+```python
+from cleveragents.skills.builtins.context_ops import (
+    build_context_skill_definition,
+)
+skill_def = build_context_skill_definition()
+```
+
 Based on ``docs/specification.md`` ACMS > CRP > ``builtin/context`` Skill.
 """
 
@@ -29,53 +40,273 @@ from __future__ import annotations
 
 from typing import Any
 
+import structlog
+from ulid import ULID
+
+from cleveragents.domain.models.core.context_fragment import (
+    ContextBudget,
+    ContextFragment,
+    FragmentProvenance,
+)
+from cleveragents.domain.models.core.skill import ResolvedToolEntry, Skill
 from cleveragents.domain.models.core.tool import ToolCapability
+from cleveragents.skills.protocol import SkillDefinition, SkillMetadata
 from cleveragents.tool.registry import ToolRegistry
 from cleveragents.tool.runtime import ToolSpec
 
+logger = structlog.get_logger(__name__)
+
 # ---------------------------------------------------------------------------
-# Handlers (stubs -- wired to future ACMS pipeline)
+# Default budget constants
+# ---------------------------------------------------------------------------
+
+_DEFAULT_MAX_TOKENS: int = 4096
+_DEFAULT_RESERVED_TOKENS: int = 512
+_MAX_QUERY_HISTORY_RESULTS: int = 50
+
+
+# ---------------------------------------------------------------------------
+# Lazy service resolution
+# ---------------------------------------------------------------------------
+
+
+def _get_pipeline() -> Any:
+    """Lazily resolve the ``ACMSPipeline`` from the DI container."""
+    from cleveragents.application.container import get_container
+
+    return get_container().acms_pipeline()
+
+
+def _get_tier_service() -> Any:
+    """Lazily resolve the ``ContextTierService`` from the DI container."""
+    from cleveragents.application.container import get_container
+
+    return get_container().context_tier_service()
+
+
+# ---------------------------------------------------------------------------
+# Fragment conversion helpers
+# ---------------------------------------------------------------------------
+
+
+def _tiered_to_pipeline_fragment(
+    tiered: Any,
+) -> ContextFragment:
+    """Convert a ``TieredFragment`` to a pipeline ``ContextFragment``.
+
+    The ACMS pipeline operates on ``ContextFragment`` (from
+    ``domain.models.core.context_fragment``), while the tier service
+    stores ``TieredFragment`` objects.  This helper bridges the two.
+    """
+    return ContextFragment(
+        fragment_id=tiered.fragment_id,
+        uko_node=tiered.resource_id or f"tier:{tiered.fragment_id}",
+        content=tiered.content,
+        detail_depth=0,
+        token_count=tiered.token_count,
+        relevance_score=0.5,
+        provenance=FragmentProvenance(
+            resource_uri=tiered.resource_id or "unknown",
+        ),
+        tier=tiered.tier.value if hasattr(tiered.tier, "value") else tiered.tier,
+    )
+
+
+def _fragment_to_dict(fragment: ContextFragment) -> dict[str, Any]:
+    """Serialise a ``ContextFragment`` to a plain dict for tool output."""
+    return {
+        "fragment_id": fragment.fragment_id,
+        "uko_node": fragment.uko_node,
+        "content": fragment.content,
+        "detail_depth": fragment.detail_depth,
+        "token_count": fragment.token_count,
+        "relevance_score": fragment.relevance_score,
+        "tier": fragment.tier,
+    }
+
+
+def _tiered_to_dict(tiered: Any) -> dict[str, Any]:
+    """Serialise a ``TieredFragment`` to a plain dict for tool output."""
+    return {
+        "fragment_id": tiered.fragment_id,
+        "content": tiered.content,
+        "tier": tiered.tier.value if hasattr(tiered.tier, "value") else tiered.tier,
+        "token_count": tiered.token_count,
+        "resource_id": tiered.resource_id,
+        "project_name": tiered.project_name,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Handlers (wired to ACMS pipeline and tier services)
 # ---------------------------------------------------------------------------
 
 
 def _handle_request_context(inputs: dict[str, Any]) -> dict[str, Any]:
-    """Request specific context to be added to the conversation.
+    """Request specific context via the ACMS pipeline.
 
-    Stub implementation -- raises ``NotImplementedError`` until the
-    ACMS Context Assembly Pipeline is available.
+    Builds a ``ContextBudget``, sources fragments from the
+    ``ContextTierService``, optionally filters by query/focus keywords,
+    and delegates to ``ACMSPipeline.assemble()`` for budget-constrained
+    context assembly.
+
+    Args:
+        inputs: Tool input dict.  Required: ``purpose``.  Optional:
+            ``query``, ``focus``, ``breadth``, ``depth``, ``plan_id``.
+
+    Returns:
+        Dict with ``fragments`` (list of serialised fragments) and
+        ``total_tokens`` (int).
     """
-    raise NotImplementedError(
-        "request_context is not yet wired to the ACMS pipeline. "
-        "This tool will be functional once the Context Assembly Pipeline "
-        "is implemented."
+    purpose: str = inputs["purpose"]
+    query: str | None = inputs.get("query")
+    focus: list[str] = inputs.get("focus", [])
+    plan_id: str = inputs.get("plan_id", str(ULID()))
+
+    pipeline = _get_pipeline()
+    tier_service = _get_tier_service()
+
+    # Source fragments from all tiers via the tier service
+    all_tiered: list[Any] = tier_service.get_all_fragments()
+
+    # Convert TieredFragments to pipeline ContextFragments
+    pipeline_fragments: list[ContextFragment] = [
+        _tiered_to_pipeline_fragment(tf) for tf in all_tiered
+    ]
+
+    # Filter by query keywords if provided
+    if query:
+        query_lower = query.lower()
+        pipeline_fragments = [
+            f
+            for f in pipeline_fragments
+            if query_lower in f.content.lower() or query_lower in f.uko_node.lower()
+        ]
+
+    # Filter by focus targets if provided
+    if focus:
+        focus_lower = [t.lower() for t in focus]
+        pipeline_fragments = [
+            f
+            for f in pipeline_fragments
+            if any(
+                t in f.content.lower() or t in f.uko_node.lower() for t in focus_lower
+            )
+        ]
+
+    budget = ContextBudget(
+        max_tokens=_DEFAULT_MAX_TOKENS,
+        reserved_tokens=_DEFAULT_RESERVED_TOKENS,
     )
 
+    # If no fragments available, return empty result
+    if not pipeline_fragments:
+        logger.info(
+            "context_ops.request_context.empty",
+            purpose=purpose,
+            query=query,
+        )
+        return {"fragments": [], "total_tokens": 0}
+
+    payload = pipeline.assemble(
+        plan_id=plan_id,
+        fragments=pipeline_fragments,
+        budget=budget,
+    )
+
+    logger.info(
+        "context_ops.request_context.assembled",
+        purpose=purpose,
+        plan_id=plan_id,
+        fragment_count=len(payload.fragments),
+        total_tokens=payload.total_tokens,
+    )
+
+    return {
+        "fragments": [_fragment_to_dict(f) for f in payload.fragments],
+        "total_tokens": payload.total_tokens,
+    }
+
 
 def _handle_query_history(inputs: dict[str, Any]) -> dict[str, Any]:
-    """Query historical context about past decisions and changes.
+    """Query historical context fragments from the tier service.
 
-    Stub implementation -- raises ``NotImplementedError`` until the
-    ACMS temporal archaeology strategy is available.
+    Searches across all tiers for fragments whose content or resource ID
+    matches the query string via case-insensitive substring matching.
+
+    Args:
+        inputs: Tool input dict. Required: ``query``.
+
+    Returns:
+        Dict with ``entries`` (list of serialised fragment dicts).
     """
-    raise NotImplementedError(
-        "query_history is not yet wired to the ACMS pipeline. "
-        "This tool will be functional once the temporal archaeology "
-        "strategy is implemented."
+    query: str = inputs["query"]
+    tier_service = _get_tier_service()
+
+    # Gather fragments from all tiers
+    all_tiered: list[Any] = tier_service.get_all_fragments()
+
+    # Filter by query keywords (case-insensitive substring)
+    query_lower = query.lower()
+    matched = [
+        tf
+        for tf in all_tiered
+        if query_lower in tf.content.lower()
+        or query_lower in tf.fragment_id.lower()
+        or query_lower in tf.resource_id.lower()
+    ]
+
+    # Sort by last_accessed descending, cap results
+    matched.sort(key=lambda f: f.last_accessed, reverse=True)
+    matched = matched[:_MAX_QUERY_HISTORY_RESULTS]
+
+    logger.info(
+        "context_ops.query_history",
+        query=query,
+        result_count=len(matched),
     )
 
+    return {"entries": [_tiered_to_dict(tf) for tf in matched]}
+
 
 def _handle_get_context_budget(inputs: dict[str, Any]) -> dict[str, Any]:
-    """Check remaining context token budget.
+    """Return the current context token budget state.
 
-    Stub implementation -- raises ``NotImplementedError`` until the
-    ACMS budget tracking is available.
+    Computes used tokens by summing token counts across all fragments
+    in the hot tier of the ``ContextTierService``.
+
+    Args:
+        inputs: Tool input dict (no required fields).
+
+    Returns:
+        Dict with ``max_tokens``, ``reserved_tokens``,
+        ``available_tokens``, and ``used_tokens``.
     """
-    raise NotImplementedError(
-        "get_context_budget is not yet wired to the ACMS pipeline. "
-        "This tool will be functional once context budget tracking "
-        "is implemented."
+    tier_service = _get_tier_service()
+
+    budget = ContextBudget(
+        max_tokens=_DEFAULT_MAX_TOKENS,
+        reserved_tokens=_DEFAULT_RESERVED_TOKENS,
     )
 
+    # Calculate used tokens from hot tier
+    used_tokens = sum(f.token_count for f in tier_service.get_hot_fragments())
+
+    logger.info(
+        "context_ops.get_context_budget",
+        max_tokens=budget.max_tokens,
+        reserved_tokens=budget.reserved_tokens,
+        available_tokens=budget.available_tokens,
+        used_tokens=used_tokens,
+    )
+
+    return {
+        "max_tokens": budget.max_tokens,
+        "reserved_tokens": budget.reserved_tokens,
+        "available_tokens": budget.available_tokens,
+        "used_tokens": used_tokens,
+    }
+
 
 # ---------------------------------------------------------------------------
 # Tool specs
@@ -117,6 +348,13 @@ SKILL_CONTEXT_REQUEST_SPEC = ToolSpec(
                 "type": "string",
                 "description": "Why do you need this context?",
             },
+            "plan_id": {
+                "type": "string",
+                "description": (
+                    "Plan identifier (ULID).  If omitted a fresh ULID is "
+                    "generated for standalone invocation."
+                ),
+            },
         },
         "required": ["purpose"],
     },
@@ -148,12 +386,6 @@ SKILL_CONTEXT_QUERY_HISTORY_SPEC = ToolSpec(
                 "type": "string",
                 "description": "What historical information do you need?",
             },
-            "scope": {
-                "type": "string",
-                "enum": ["current_plan", "plan_tree", "all_plans"],
-                "default": "plan_tree",
-                "description": "Scope of history to search",
-            },
         },
         "required": ["query"],
     },
@@ -210,7 +442,79 @@ ALL_SKILL_CONTEXT_TOOLS: list[ToolSpec] = [
 ]
 
 
+# ---------------------------------------------------------------------------
+# Tool registration
+# ---------------------------------------------------------------------------
+
+
 def register_skill_context_tools(registry: ToolRegistry) -> None:
     """Register all skill-level context tools into *registry*."""
     for spec in ALL_SKILL_CONTEXT_TOOLS:
         registry.register(spec)
+
+
+# ---------------------------------------------------------------------------
+# SkillDefinition factory
+# ---------------------------------------------------------------------------
+
+#: Canonical skill name for the builtin context skill.
+BUILTIN_CONTEXT_SKILL_NAME: str = "builtin/context"
+
+
+def build_context_skill_definition() -> SkillDefinition:
+    """Build a ``SkillDefinition`` for the ``builtin/context`` skill.
+
+    The returned definition wraps the three CRP tools
+    (``request_context``, ``query_history``, ``get_context_budget``)
+    and can be registered in the ``SkillRegistry`` at startup.
+
+    Returns:
+        A frozen ``SkillDefinition`` ready for registry insertion.
+    """
+    skill = Skill(
+        name=BUILTIN_CONTEXT_SKILL_NAME,
+        description=(
+            "Context Request Protocol (CRP) skill providing tools for "
+            "requesting context, querying history, and checking budget "
+            "during plan execution."
+        ),
+        tool_refs=[
+            "builtin/request-context",
+            "builtin/query-history",
+            "builtin/get-context-budget",
+        ],
+    )
+
+    resolved_tools = [
+        ResolvedToolEntry(
+            name="builtin/request-context",
+            source_skill=BUILTIN_CONTEXT_SKILL_NAME,
+            is_inline=False,
+        ),
+        ResolvedToolEntry(
+            name="builtin/query-history",
+            source_skill=BUILTIN_CONTEXT_SKILL_NAME,
+            is_inline=False,
+        ),
+        ResolvedToolEntry(
+            name="builtin/get-context-budget",
+            source_skill=BUILTIN_CONTEXT_SKILL_NAME,
+            is_inline=False,
+        ),
+    ]
+
+    metadata = SkillMetadata(
+        name=BUILTIN_CONTEXT_SKILL_NAME,
+        description=skill.description,
+        version="1.0.0",
+        tool_count=len(resolved_tools),
+        source_types=["tool_ref"],
+        writes=False,
+        read_only=True,
+    )
+
+    return SkillDefinition(
+        skill=skill,
+        resolved_tools=resolved_tools,
+        metadata=metadata,
+    )