CHANGELOG.md

@@ -6,6 +6,7 @@ Changed `wf10_batch.robot` to be less likely to create files, and
 `plan_generation_graph.robot` to give more test answers.
 
 ## [Unreleased]
+- **feat(context): PriorityContextStrategy** (#9997 / PR #10772): Implements a priority-based context strategy that ranks context fragments by configurable priority scores — default role-based rules (system > tool > user > assistant), exponential recency decay (7-day half-life), and explicit priority tag boost. Supports custom scoring function injection and custom PriorityRule list injection. Registered in the ACMS pipeline under key `priority_context`. `PriorityRule` uses Pydantic `BaseModel` for architecture conformance. Includes 18 BDD scenarios covering all acceptance criteria.
 - **docs(timeline): verify timeline status for 2026-04-16 Cycle 2** (#8519): Updated `docs/timeline.md` with Days 104-106 Cycle 2 milestone snapshot. No changes detected since Cycle 1. All M3-M7 milestones remain overdue. Timeline verification performed by AUTO-TIME-3 supervisor agent. Refs #8519.
 - **fix(cli/plan): plan correct JSON output envelope fix and BDD test coverage** (#8584 / PR #8662): Restructured `agents plan correct --format json` output to nest correction fields under `data.correction` (e.g., `data.correction.mode`) and populate the spec-required CLI envelope with `command="plan correct"`, `status`, `exit_code`, `timing`, and `messages` fields. Added three BDD scenarios in `features/tdd_plan_correct_json_output.feature` validating the envelope structure for both revert and append modes.
 - **fix(cli): add --url flag to resource add for git resource type** (#6322): Added support for the `--url` flag on `agents resource add git` command, allowing users to specify a remote URL for git resources. The flag is validated to only apply to git resource types. Includes Behave BDD tests in `features/resource_cli_git_url_flag.feature` and Robot Framework integration tests verifying correct URL validation and CLI behavior.

features/priority_context_strategy.feature

@@ -0,0 +1,168 @@
+@phase2 @acms @context_strategies @priority_context
+Feature: PriorityContextStrategy with configurable priority scoring
+  As a CleverAgents developer
+  I want a priority-based context strategy
+  So that high-priority messages are always retained when the context window must be trimmed
+
+  @priority_context_protocol
+  Scenario: PriorityContextStrategy satisfies ContextStrategy protocol
+    Given a PriorityContextStrategy with default rules
+    Then the PriorityContextStrategy should satisfy the ContextStrategy protocol
+    And the PriorityContextStrategy name should be "priority_context"
+
+  @priority_context_protocol
+  Scenario: PriorityContextStrategy reports capabilities
+    Given a PriorityContextStrategy with default rules
+    Then the PriorityContextStrategy capabilities should be a StrategyCapabilities instance
+
+  @priority_context_protocol
+  Scenario: PriorityContextStrategy explain returns non-empty description
+    Given a PriorityContextStrategy with default rules
+    Then the PriorityContextStrategy explain should contain "priority"
+
+  @priority_context_can_handle
+  Scenario: PriorityContextStrategy can_handle returns 0.75 with any request
+    Given a PriorityContextStrategy with default rules
+    When I check can_handle on PriorityContextStrategy with query "test"
+    Then the strategy confidence should be 0.75
+
+  @priority_context_can_handle
+  Scenario: PriorityContextStrategy can_handle returns 0.75 without query
+    Given a PriorityContextStrategy with default rules
+    When I check can_handle on PriorityContextStrategy without query
+    Then the strategy confidence should be 0.75
+
+  @priority_context_role
+  Scenario: System role fragments are ranked highest by default
+    Given a PriorityContextStrategy with default rules
+    And the following priority strategy fragments:
+      | uko_node              | content                  | score | tokens | depth | role      |
+      | project://app/a.py    | assistant message        | 0.9   | 20     | 3     | assistant |
+      | project://app/b.py    | system instruction       | 0.5   | 20     | 3     | system    |
+      | project://app/c.py    | user request             | 0.7   | 20     | 3     | user      |
+      | project://app/d.py    | tool result              | 0.6   | 20     | 3     | tool      |
+    And a strategy budget with max_tokens 1000 and reserved_tokens 0
+    When I assemble with the PriorityContextStrategy
+    Then the first result fragment should have uko_node "project://app/b.py"
+
+  @priority_context_role
+  Scenario: Tool role fragments are ranked above user and assistant
+    Given a PriorityContextStrategy with default rules
+    And the following priority strategy fragments:
+      | uko_node              | content                  | score | tokens | depth | role      |
+      | project://app/a.py    | assistant message        | 0.9   | 20     | 3     | assistant |
+      | project://app/b.py    | user request             | 0.8   | 20     | 3     | user      |
+      | project://app/c.py    | tool result              | 0.5   | 20     | 3     | tool      |
+    And a strategy budget with max_tokens 1000 and reserved_tokens 0
+    When I assemble with the PriorityContextStrategy
+    Then the first result fragment should have uko_node "project://app/c.py"
+
+  @priority_context_role
+  Scenario: User role fragments are ranked above assistant
+    Given a PriorityContextStrategy with default rules
+    And the following priority strategy fragments:
+      | uko_node              | content                  | score | tokens | depth | role      |
+      | project://app/a.py    | assistant message        | 0.9   | 20     | 3     | assistant |
+      | project://app/b.py    | user request             | 0.5   | 20     | 3     | user      |
+    And a strategy budget with max_tokens 1000 and reserved_tokens 0
+    When I assemble with the PriorityContextStrategy
+    Then the first result fragment should have uko_node "project://app/b.py"
+
+  @priority_context_tag
+  Scenario: Fragments with priority tag are boosted above role-based score
+    Given a PriorityContextStrategy with default rules
+    And the following priority strategy fragments with tags:
+      | uko_node              | content                  | score | tokens | depth | role      | priority_tag |
+      | project://app/a.py    | system instruction       | 0.5   | 20     | 3     | system    |              |
+      | project://app/b.py    | user request             | 0.5   | 20     | 3     | user      | high         |
+    And a strategy budget with max_tokens 1000 and reserved_tokens 0
+    When I assemble with the PriorityContextStrategy
+    Then the first result fragment should have uko_node "project://app/b.py"
+
+  @priority_context_tag
+  Scenario: Fragments without priority tag are not boosted
+    Given a PriorityContextStrategy with default rules
+    And the following priority strategy fragments with tags:
+      | uko_node              | content                  | score | tokens | depth | role      | priority_tag |
+      | project://app/a.py    | system instruction       | 0.9   | 20     | 3     | system    |              |
+      | project://app/b.py    | user request             | 0.5   | 20     | 3     | user      |              |
+    And a strategy budget with max_tokens 1000 and reserved_tokens 0
+    When I assemble with the PriorityContextStrategy
+    Then the first result fragment should have uko_node "project://app/a.py"
+
+  @priority_context_budget
+  Scenario: PriorityContextStrategy respects token budget
+    Given a PriorityContextStrategy with default rules
+    And the following priority strategy fragments:
+      | uko_node              | content                  | score | tokens | depth | role   |
+      | project://app/a.py    | system instruction one   | 0.9   | 100    | 3     | system |
+      | project://app/b.py    | system instruction two   | 0.8   | 100    | 3     | system |
+      | project://app/c.py    | system instruction three | 0.7   | 100    | 3     | system |
+    And a strategy budget with max_tokens 250 and reserved_tokens 0
+    When I assemble with the PriorityContextStrategy
+    Then 2 fragments should be returned by priority strategy
+
+  @priority_context_budget
+  Scenario: PriorityContextStrategy returns empty for empty input
+    Given a PriorityContextStrategy with default rules
+    And an empty priority strategy fragment list
+    And a strategy budget with max_tokens 1000 and reserved_tokens 0
+    When I assemble with the PriorityContextStrategy
+    Then 0 fragments should be returned by priority strategy
+
+  @priority_context_custom_fn
+  Scenario: Custom scoring function overrides default scoring
+    Given a PriorityContextStrategy with a custom score function that boosts depth
+    And the following priority strategy fragments:
+      | uko_node              | content                  | score | tokens | depth | role      |
+      | project://app/a.py    | system instruction       | 0.9   | 20     | 1     | system    |
+      | project://app/b.py    | user request             | 0.5   | 20     | 9     | user      |
+    And a strategy budget with max_tokens 1000 and reserved_tokens 0
+    When I assemble with the PriorityContextStrategy
+    Then the first result fragment should have uko_node "project://app/b.py"
+
+  @priority_context_custom_fn
+  Scenario: Custom scoring function receives fragment and returns float
+    Given a PriorityContextStrategy with a custom score function that returns constant 1.0
+    And the following priority strategy fragments:
+      | uko_node              | content                  | score | tokens | depth | role      |
+      | project://app/a.py    | system instruction       | 0.9   | 20     | 3     | system    |
+      | project://app/b.py    | user request             | 0.5   | 20     | 3     | user      |
+    And a strategy budget with max_tokens 1000 and reserved_tokens 0
+    When I assemble with the PriorityContextStrategy
+    Then 2 fragments should be returned by priority strategy
+
+  @priority_context_rule
+  Scenario: PriorityRule has field, matcher, and score attributes
+    Given a PriorityRule with field "role" matcher "system" and score 1.0
+    Then the PriorityRule field should be "role"
+    And the PriorityRule matcher should be "system"
+    And the PriorityRule score should be 1.0
+
+  @priority_context_rule
+  Scenario: PriorityContextStrategy accepts custom PriorityRule list
+    Given a PriorityContextStrategy with custom rules boosting "tool" role to 2.0
+    And the following priority strategy fragments:
+      | uko_node              | content                  | score | tokens | depth | role      |
+      | project://app/a.py    | system instruction       | 0.9   | 20     | 3     | system    |
+      | project://app/b.py    | tool result              | 0.5   | 20     | 3     | tool      |
+    And a strategy budget with max_tokens 1000 and reserved_tokens 0
+    When I assemble with the PriorityContextStrategy
+    Then the first result fragment should have uko_node "project://app/b.py"
+
+  @priority_context_registry
+  Scenario: PriorityContextStrategy is registered in plugin registry under "priority_context"
+    Given an ACMS pipeline for strategy tests
+    When I register PriorityContextStrategy with the pipeline
+    Then the pipeline should have strategy "priority_context"
+
+  @priority_context_recency
+  Scenario: More recent fragments score higher with recency decay rule
+    Given a PriorityContextStrategy with default rules
+    And the following priority strategy fragments with recency:
+      | uko_node              | content                  | score | tokens | depth | role | days_old |
+      | project://app/a.py    | old system instruction   | 0.9   | 20     | 3     | user | 30       |
+      | project://app/b.py    | recent user request      | 0.5   | 20     | 3     | user | 0        |
+    And a strategy budget with max_tokens 1000 and reserved_tokens 0
+    When I assemble with the PriorityContextStrategy
+    Then the first result fragment should have uko_node "project://app/b.py"

features/steps/priority_context_strategy_steps.py

@@ -0,0 +1,298 @@
+"""Step definitions for ``features/priority_context_strategy.feature``.
+
+Covers the PriorityContextStrategy with configurable priority scoring:
+
+* Protocol conformance
+* can_handle confidence
+* Role-based priority rules (system > tool > user > assistant)
+* Explicit priority tag boost
+* Budget enforcement
+* Custom scoring function injection
+* PriorityRule dataclass
+* Plugin registry registration
+* Recency decay rule
+
+Note: Steps shared with other strategy tests (e.g. ``the strategy confidence
+should be``, ``a strategy budget with max_tokens``, ``an ACMS pipeline for
+strategy tests``, ``the pipeline should have strategy``, and ``the first
+result fragment should have uko_node``) are defined in
+``context_strategies_steps.py`` and reused here.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.application.services.acms_service import (
+    ContextStrategy,
+    StrategyCapabilities,
+)
+from cleveragents.application.services.priority_context_strategy import (
+    PriorityContextStrategy,
+    PriorityRule,
+)
+from cleveragents.domain.models.core.context_fragment import (
+    ContextFragment,
+    FragmentProvenance,
+)
+
+__all__: list[str] = []
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_priority_fragment(
+    uko_node: str,
+    content: str,
+    score: float,
+    tokens: int,
+    depth: int,
+    role: str = "",
+    priority_tag: str = "",
+    created_at: datetime | None = None,
+) -> ContextFragment:
+    """Build a ``ContextFragment`` for priority strategy tests."""
+    metadata: dict[str, str] = {}
+    if role:
+        metadata["role"] = role
+    if priority_tag:
+        metadata["priority_tag"] = priority_tag
+
+    kwargs: dict[str, Any] = {
+        "uko_node": uko_node,
+        "content": content,
+        "relevance_score": score,
+        "token_count": tokens,
+        "detail_depth": depth,
+        "provenance": FragmentProvenance(resource_uri=uko_node),
+        "metadata": metadata,
+    }
+    if created_at is not None:
+        kwargs["created_at"] = created_at
+
+    return ContextFragment(**kwargs)
+
+
+# ---------------------------------------------------------------------------
+# Given steps — strategy construction
+# ---------------------------------------------------------------------------
+
+
+@given("a PriorityContextStrategy with default rules")
+def step_priority_strategy_default(context: Context) -> None:
+    context.strategy = PriorityContextStrategy()
+
+
+@given("a PriorityContextStrategy with a custom score function that boosts depth")
+def step_priority_strategy_custom_depth(context: Context) -> None:
+    def depth_score(frag: ContextFragment) -> float:
+        return frag.detail_depth / 9.0
+
+    context.strategy = PriorityContextStrategy(score_fn=depth_score)
+
+
+@given(
+    "a PriorityContextStrategy with a custom score function that returns constant 1.0"
+)
+def step_priority_strategy_custom_constant(context: Context) -> None:
+    def constant_score(frag: ContextFragment) -> float:
+        return 1.0
+
+    context.strategy = PriorityContextStrategy(score_fn=constant_score)
+
+
+@given(
+    'a PriorityContextStrategy with custom rules boosting "{role}" role to {boost:g}'
+)
+def step_priority_strategy_custom_rules(
+    context: Context, role: str, boost: float
+) -> None:
+    rules = [PriorityRule(field="role", matcher=role, score=boost)]
+    context.strategy = PriorityContextStrategy(rules=rules)
+
+
+# ---------------------------------------------------------------------------
+# Given steps — PriorityRule construction
+# ---------------------------------------------------------------------------
+
+
+@given('a PriorityRule with field "{field}" matcher "{matcher}" and score {score:g}')
+def step_priority_rule_construct(
+    context: Context, field: str, matcher: str, score: float
+) -> None:
+    context.priority_rule = PriorityRule(field=field, matcher=matcher, score=score)
+
+
+# ---------------------------------------------------------------------------
+# Given steps — fragment tables
+# ---------------------------------------------------------------------------
+
+
+@given("the following priority strategy fragments:")
+def step_priority_fragments_table(context: Context) -> None:
+    context.priority_fragments = []
+    for row in context.table:
+        role = row.get("role", "")
+        frag = _make_priority_fragment(
+            uko_node=row["uko_node"],
+            content=row["content"],
+            score=float(row["score"]),
+            tokens=int(row["tokens"]),
+            depth=int(row["depth"]),
+            role=role,
+        )
+        context.priority_fragments.append(frag)
+
+
+@given("the following priority strategy fragments with tags:")
+def step_priority_fragments_with_tags_table(context: Context) -> None:
+    context.priority_fragments = []
+    for row in context.table:
+        role = row.get("role", "")
+        priority_tag = row.get("priority_tag", "")
+        frag = _make_priority_fragment(
+            uko_node=row["uko_node"],
+            content=row["content"],
+            score=float(row["score"]),
+            tokens=int(row["tokens"]),
+            depth=int(row["depth"]),
+            role=role,
+            priority_tag=priority_tag,
+        )
+        context.priority_fragments.append(frag)
+
+
+@given("the following priority strategy fragments with recency:")
+def step_priority_fragments_with_recency_table(context: Context) -> None:
+    context.priority_fragments = []
+    now = datetime.now(UTC)
+    for row in context.table:
+        role = row.get("role", "")
+        days_old = int(row.get("days_old", "0"))
+        created_at = now - timedelta(days=days_old)
+        frag = _make_priority_fragment(
+            uko_node=row["uko_node"],
+            content=row["content"],
+            score=float(row["score"]),
+            tokens=int(row["tokens"]),
+            depth=int(row["depth"]),
+            role=role,
+            created_at=created_at,
+        )
+        context.priority_fragments.append(frag)
+
+
+@given("an empty priority strategy fragment list")
+def step_empty_priority_fragments(context: Context) -> None:
+    context.priority_fragments = []
+
+
+# ---------------------------------------------------------------------------
+# When steps
+# ---------------------------------------------------------------------------
+
+
+@when("I assemble with the PriorityContextStrategy")
+def step_assemble_priority(context: Context) -> None:
+    result = list(
+        context.strategy.assemble(context.priority_fragments, context.strategy_budget)
+    )
+    # Set both context attributes so shared steps from context_strategies_steps.py
+    # (which use context.strategy_result) work correctly.
+    context.strategy_result = result
+    context.priority_result = result
+
+
+@when('I check can_handle on PriorityContextStrategy with query "{query}"')
+def step_can_handle_priority_with_query(context: Context, query: str) -> None:
+    request: dict[str, Any] = {"query": query}
+    context.confidence = context.strategy.can_handle(request)
+
+
+@when("I check can_handle on PriorityContextStrategy without query")
+def step_can_handle_priority_no_query(context: Context) -> None:
+    request: dict[str, Any] = {}
+    context.confidence = context.strategy.can_handle(request)
+
+
+@when("I register PriorityContextStrategy with the pipeline")
+def step_register_priority_strategy(context: Context) -> None:
+    context.pipeline.register_strategy("priority_context", PriorityContextStrategy())
+
+
+# ---------------------------------------------------------------------------
+# Then steps — protocol conformance
+# ---------------------------------------------------------------------------
+
+
+@then("the PriorityContextStrategy should satisfy the ContextStrategy protocol")
+def step_priority_satisfies_protocol(context: Context) -> None:
+    assert isinstance(context.strategy, ContextStrategy), (
+        "PriorityContextStrategy does not satisfy the ContextStrategy protocol"
+    )
+
+
+@then('the PriorityContextStrategy name should be "{name}"')
+def step_priority_name(context: Context, name: str) -> None:
+    actual = context.strategy.name
+    assert actual == name, f"Expected name '{name}', got '{actual}'"
+
+
+@then(
+    "the PriorityContextStrategy capabilities should be a StrategyCapabilities instance"
+)
+def step_priority_capabilities(context: Context) -> None:
+    caps = context.strategy.capabilities
+    assert isinstance(caps, StrategyCapabilities), (
+        f"Expected StrategyCapabilities, got {type(caps).__name__}"
+    )
+
+
+@then('the PriorityContextStrategy explain should contain "{text}"')
+def step_priority_explain(context: Context, text: str) -> None:
+    explanation = context.strategy.explain()
+    assert text.lower() in explanation.lower(), (
+        f"Expected explain to contain '{text}', got: {explanation}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Then steps — results
+# ---------------------------------------------------------------------------
+
+
+@then("{count:d} fragments should be returned by priority strategy")
+def step_priority_fragment_count(context: Context, count: int) -> None:
+    result = getattr(context, "priority_result", [])
+    actual = len(result)
+    assert actual == count, f"Expected {count} fragments, got {actual}"
+
+
+# ---------------------------------------------------------------------------
+# Then steps — PriorityRule
+# ---------------------------------------------------------------------------
+
+
+@then('the PriorityRule field should be "{field}"')
+def step_priority_rule_field(context: Context, field: str) -> None:
+    actual = context.priority_rule.field
+    assert actual == field, f"Expected field '{field}', got '{actual}'"
+
+
+@then('the PriorityRule matcher should be "{matcher}"')
+def step_priority_rule_matcher(context: Context, matcher: str) -> None:
+    actual = context.priority_rule.matcher
+    assert actual == matcher, f"Expected matcher '{matcher}', got '{actual}'"
+
+
+@then("the PriorityRule score should be {score:g}")
+def step_priority_rule_score(context: Context, score: float) -> None:
+    actual = context.priority_rule.score
+    assert abs(actual - score) < 1e-6, f"Expected score {score}, got {actual}"

src/cleveragents/application/services/acms_service.py

@@ -74,6 +74,22 @@ def _get_semantic_chunking_strategy_class() -> type:
     return _SemanticChunkingStrategy
 
 
+# Lazy import helper for PriorityContextStrategy (issue #9997).
+_PriorityContextStrategy: type | None = None
+
+
+def _get_priority_context_strategy_class() -> type:
+    """Return the :class:`PriorityContextStrategy` class, importing lazily."""
+    global _PriorityContextStrategy
+    if _PriorityContextStrategy is None:
+        from cleveragents.application.services.priority_context_strategy import (
+            PriorityContextStrategy,
+        )
+
+        _PriorityContextStrategy = PriorityContextStrategy
+    return _PriorityContextStrategy
+
+
 _SPEC_BUILTIN_STRATEGIES: dict[str, Any] | None = None
 
 
@@ -815,6 +831,10 @@ class ACMSPipeline:
         if "semantic_chunking" not in self._strategies:
             _sc_cls = _get_semantic_chunking_strategy_class()
             self._strategies["semantic_chunking"] = cast(ContextStrategy, _sc_cls())
+        # Register PriorityContextStrategy (issue #9997).
+        if "priority_context" not in self._strategies:
+            _pc_cls = _get_priority_context_strategy_class()
+            self._strategies["priority_context"] = cast(ContextStrategy, _pc_cls())
         # Register the 6 spec-required built-in strategies via adapters.
         # These strategies implement the domain-model ContextStrategy protocol
         # (strategy_stubs.py) and are wrapped in SpecStrategyAdapter instances

src/cleveragents/application/services/priority_context_strategy.py

@@ -0,0 +1,285 @@
+"""PriorityContextStrategy — priority-based context strategy.
+
+Implements a context strategy that assigns configurable priority scores to
+fragments based on message role, recency, and explicit priority tags.
+High-priority fragments are always retained when the context window must be
+trimmed, improving agent reliability in constrained environments.
+
+The strategy supports:
+
+1. **Role-based priority rules** — built-in rules that score fragments by
+   message role: system > tool > user > assistant.
+2. **Recency decay** — more recent fragments receive a higher score.
+3. **Explicit priority tag boost** — fragments with a ``priority_tag``
+   metadata key receive an additional score boost.
+4. **Custom scoring function injection** — callers may supply a
+   ``score_fn: Callable[[ContextFragment], float]`` that completely
+   overrides the default scoring pipeline.
+5. **Custom PriorityRule list** — callers may supply a list of
+   ``PriorityRule`` dataclasses to replace the default role-based rules.
+
+The strategy is registered in the ACMS pipeline under the key
+``"priority_context"``.
+
+Based on issue #9997 acceptance criteria.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable, Sequence
+from datetime import UTC, datetime
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from cleveragents.application.services.acms_service import (
+    StrategyCapabilities,
+    _pack_budget,
+)
+from cleveragents.domain.models.core.context_fragment import (
+    ContextBudget,
+    ContextFragment,
+)
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Role priority constants (system > tool > user > assistant)
+# ---------------------------------------------------------------------------
+
+_ROLE_SCORES: dict[str, float] = {
+    "system": 1.0,
+    "tool": 0.75,
+    "user": 0.5,
+    "assistant": 0.25,
+}
+
+# Boost applied to fragments with a non-empty ``priority_tag`` metadata key.
+# Must be large enough to override the highest role-based score (1.0) when
+# combined with a lower-priority role score.  Set to 0.6 so that a ``user``
+# role fragment with a priority tag (0.5 + 0.6 = 1.1) outranks a ``system``
+# role fragment without a tag (1.0).
+_PRIORITY_TAG_BOOST: float = 0.6
+
+# Maximum recency decay contribution (applied to the most recent fragment).
+_MAX_RECENCY_SCORE: float = 0.3
+
+# Recency half-life in days: score halves every this many days.
+_RECENCY_HALF_LIFE_DAYS: float = 7.0
+
+
+# ---------------------------------------------------------------------------
+# PriorityRule model
+# ---------------------------------------------------------------------------
+
+
+class PriorityRule(BaseModel):
+    """A single priority scoring rule.
+
+    Attributes:
+        field:   The metadata field name to inspect (e.g. ``"role"``).
+        matcher: The value to match against the field (e.g. ``"system"``).
+        score:   The priority score to assign when the field matches.
+    """
+
+    field: str = Field(..., description="Metadata field name to inspect")
+    matcher: str = Field(..., description="Value to match against the field")
+    score: float = Field(..., description="Priority score when the field matches")
+
+    model_config = ConfigDict(str_strip_whitespace=True, validate_assignment=True)
+
+
+# ---------------------------------------------------------------------------
+# Default built-in rules
+# ---------------------------------------------------------------------------
+
+DEFAULT_PRIORITY_RULES: list[PriorityRule] = [
+    PriorityRule(field="role", matcher="system", score=_ROLE_SCORES["system"]),
+    PriorityRule(field="role", matcher="tool", score=_ROLE_SCORES["tool"]),
+    PriorityRule(field="role", matcher="user", score=_ROLE_SCORES["user"]),
+    PriorityRule(field="role", matcher="assistant", score=_ROLE_SCORES["assistant"]),
+]
+
+
+# ---------------------------------------------------------------------------
+# PriorityContextStrategy
+# ---------------------------------------------------------------------------
+
+
+class PriorityContextStrategy:
+    """Priority-based context strategy with configurable scoring rules.
+
+    Assigns a composite priority score to each fragment based on:
+
+    1. **Role-based rules** — ``PriorityRule`` list matching metadata fields.
+    2. **Recency decay** — more recent fragments score higher.
+    3. **Explicit priority tag boost** — fragments with a non-empty
+       ``priority_tag`` metadata key receive an additional boost.
+
+    When a custom ``score_fn`` is provided it completely replaces the
+    default scoring pipeline.  When a custom ``rules`` list is provided
+    it replaces the default role-based rules (recency and tag boost still
+    apply unless ``score_fn`` is also provided).
+
+    Implements the ``ContextStrategy`` protocol from ``acms_service.py``.
+
+    Registered in the ACMS pipeline under key ``"priority_context"``.
+    """
+
+    def __init__(
+        self,
+        *,
+        rules: list[PriorityRule] | None = None,
+        score_fn: Callable[[ContextFragment], float] | None = None,
+    ) -> None:
+        """Initialise the strategy.
+
+        Args:
+            rules:    Custom priority rules.  Defaults to
+                      ``DEFAULT_PRIORITY_RULES`` when ``None``.
+            score_fn: Optional custom scoring function.  When provided it
+                      completely overrides the default scoring pipeline.
+                      The function receives a ``ContextFragment`` and must
+                      return a ``float`` priority score.
+        """
+        self._rules: list[PriorityRule] = (
+            rules if rules is not None else list(DEFAULT_PRIORITY_RULES)
+        )
+        self._score_fn: Callable[[ContextFragment], float] | None = score_fn
+
+    @property
+    def name(self) -> str:
+        return "priority_context"
+
+    @property
+    def capabilities(self) -> StrategyCapabilities:
+        return StrategyCapabilities(
+            supports_semantic_search=False,
+            supports_graph_navigation=False,
+            supports_temporal_archaeology=False,
+        )
+
+    def can_handle(self, request: dict[str, Any]) -> float:
+        """Return 0.75 — high confidence for any request.
+
+        The priority strategy is a general-purpose strategy that can
+        handle any request regardless of backend availability.
+        """
+        return 0.75
+
+    def assemble(
+        self,
+        fragments: Sequence[ContextFragment],
+        budget: ContextBudget,
+    ) -> Sequence[ContextFragment]:
+        """Rank fragments by priority score and pack within budget.
+
+        Fragments are scored using the configured scoring pipeline
+        (custom ``score_fn`` or default rule-based scoring), then sorted
+        in descending order and packed greedily within the token budget.
+
+        Args:
+            fragments: Input fragments to rank.
+            budget:    Token budget constraint.
+
+        Returns:
+            Highest-scoring fragments that fit within the budget.
+        """
+        if not fragments:
+            return list(fragments)
+
+        scored: list[tuple[ContextFragment, float]] = []
+        for frag in fragments:
+            score = self._compute_score(frag)
+            scored.append((frag, score))
+
+        scored.sort(key=lambda pair: pair[1], reverse=True)
+        sorted_frags = [frag for frag, _ in scored]
+
+        logger.info(
+            "PriorityContext ranked fragments",
+            extra={
+                "fragment_count": len(fragments),
+                "has_custom_fn": self._score_fn is not None,
+                "rule_count": len(self._rules),
+            },
+        )
+        return _pack_budget(sorted_frags, budget)
+
+    def explain(self) -> str:
+        return (
+            "Priority-based context strategy. Assigns configurable priority "
+            "scores to fragments based on message role (system > tool > user "
+            "> assistant), recency decay, and explicit priority tags. "
+            "High-priority fragments are always retained within the token "
+            "budget. Supports custom scoring function injection and custom "
+            "PriorityRule lists. Quality 0.75."
+        )
+
+    # ------------------------------------------------------------------
+    # Internal scoring helpers
+    # ------------------------------------------------------------------
+
+    def _compute_score(self, frag: ContextFragment) -> float:
+        """Compute the priority score for a single fragment.
+
+        When a custom ``score_fn`` is configured it is called directly.
+        Otherwise the default pipeline is used:
+
+        1. Apply matching ``PriorityRule`` scores (highest match wins).
+        2. Add recency decay contribution.
+        3. Add explicit priority tag boost.
+        """
+        if self._score_fn is not None:
+            return self._score_fn(frag)
+
+        # Step 1: Rule-based score (highest matching rule wins)
+        rule_score = self._apply_rules(frag)
+
+        # Step 2: Recency decay
+        recency_score = self._recency_score(frag)
+
+        # Step 3: Priority tag boost
+        tag_boost = self._tag_boost(frag)
+
+        return rule_score + recency_score + tag_boost
+
+    def _apply_rules(self, frag: ContextFragment) -> float:
+        """Return the highest matching rule score for the fragment."""
+        best: float = 0.0
+        for rule in self._rules:
+            field_value = frag.metadata.get(rule.field, "")
+            if field_value == rule.matcher and rule.score > best:
+                best = rule.score
+        return best
+
+    @staticmethod
+    def _recency_score(frag: ContextFragment) -> float:
+        """Compute a recency contribution using exponential decay.
+
+        Returns a value in ``[0.0, _MAX_RECENCY_SCORE]``.  The most
+        recently created fragment receives the maximum score; older
+        fragments decay exponentially with a half-life of
+        ``_RECENCY_HALF_LIFE_DAYS`` days.
+        """
+        now = datetime.now(UTC)
+        created = frag.created_at
+        # ContextFragment.created_at defaults to ``datetime.now(UTC)`` — always
+        # tz-aware. The naive-datetime branch is a defensive guard for callers
+        # that construct fragments with a naive ``created_at``; not exercised
+        # by current tests.
+        if created.tzinfo is None:  # pragma: no cover
+            created = created.replace(tzinfo=UTC)
+        age_days = max(0.0, (now - created).total_seconds() / 86400.0)
+        # Exponential decay: score = max_score * 0.5^(age / half_life)
+        decay = 0.5 ** (age_days / _RECENCY_HALF_LIFE_DAYS)
+        return _MAX_RECENCY_SCORE * decay
+
+    @staticmethod
+    def _tag_boost(frag: ContextFragment) -> float:
+        """Return the priority tag boost if the fragment has a priority tag."""
+        tag = frag.metadata.get("priority_tag", "")
+        if tag:
+            return _PRIORITY_TAG_BOOST
+        return 0.0