上下文领域：建立有界卡片检索、预算与安全游标模型

新增 ContextRequest、ContextCard、ContextBundle、ContextPage、ContextSufficiency 和 ContextReceipt。实现中文词法排序、内容去重、来源多样性、最终响应预算、超大 Card outline 降级和签名分页游标；覆盖重复内容有限遍历、篡改游标、token 缩页及超大首项恢复测试。

script_build_host/src/script_build_host/domain/__init__.py

@@ -8,6 +8,14 @@ from .artifacts import (
     DirectionPreference,
     EvidenceRecordV1,
 )
+from .context_broker import (
+    ContextBundle,
+    ContextCard,
+    ContextPage,
+    ContextReceipt,
+    ContextRequest,
+    ContextSufficiency,
+)
 from .errors import ScriptBuildError
 from .goal_coverage import GoalCoverage
 from .input_snapshot import ScriptBuildInput, ScriptBuildInputSnapshotV1
@@ -18,6 +26,12 @@ __all__ = [
     "ArtifactState",
     "ArtifactVersion",
     "BuildStatus",
+    "ContextBundle",
+    "ContextCard",
+    "ContextPage",
+    "ContextReceipt",
+    "ContextRequest",
+    "ContextSufficiency",
     "DirectionArtifact",
     "DirectionConstraint",
     "DirectionGoal",

script_build_host/src/script_build_host/domain/context_broker.py

@@ -0,0 +1,560 @@
+"""Pure Context Broker models, ranking, budgeting, and continuation cursors."""
+
+from __future__ import annotations
+
+import base64
+import hmac
+import json
+import math
+import re
+from collections import Counter
+from collections.abc import Mapping, Sequence
+from dataclasses import asdict, dataclass, field, replace
+from hashlib import sha256
+from typing import Any, Literal
+
+ContextStatus = Literal["sufficient", "ambiguous", "insufficient"]
+DetailLevel = Literal["summary", "semantic", "full"]
+
+_ASCII_TERM = re.compile(r"[A-Za-z0-9]+")
+_CJK_RUN = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff\uf900-\ufaff]+")
+_CURSOR_VERSION = 1
+_PAGE_ENVELOPE_TOKEN_RESERVE = 500
+
+
+@dataclass(frozen=True, slots=True)
+class ContextRequest:
+    """A protected, bounded request passed from the Host to the Broker."""
+
+    root_trace_id: str
+    role: str
+    view: str
+    task_id: str | None = None
+    task_kind: str | None = None
+    goal_ids: tuple[str, ...] = ()
+    scope_selector: Mapping[str, Any] | None = None
+    query: str = ""
+    source_types: tuple[str, ...] = ()
+    known_revision: str | None = None
+    cursor: str | None = None
+    detail_level: DetailLevel = "summary"
+    page_size: int = 8
+    token_budget: int = 2_000
+
+    def __post_init__(self) -> None:
+        if not self.root_trace_id.strip() or not self.role.strip() or not self.view.strip():
+            raise ValueError("root_trace_id, role, and view are required")
+        if not 1 <= self.page_size <= 20:
+            raise ValueError("page_size must be between 1 and 20")
+        if self.token_budget <= 0:
+            raise ValueError("token_budget must be positive")
+        if len(set(self.goal_ids)) != len(self.goal_ids):
+            raise ValueError("goal_ids must be unique")
+        if len(set(self.source_types)) != len(self.source_types):
+            raise ValueError("source_types must be unique")
+
+    def filter_digest(self) -> str:
+        value = {
+            "role": self.role,
+            "task_id": self.task_id,
+            "task_kind": self.task_kind,
+            "goal_ids": self.goal_ids,
+            "scope_selector": self.scope_selector,
+            "query": self.query.strip(),
+            "source_types": self.source_types,
+            "detail_level": self.detail_level,
+            "page_size": self.page_size,
+            "token_budget": self.token_budget,
+        }
+        return canonical_digest(value)
+
+
+@dataclass(frozen=True, slots=True)
+class ContextCard:
+    handle: str
+    source_type: str
+    summary: str
+    excerpt: str = ""
+    goal_ids: tuple[str, ...] = ()
+    scope_selector: Mapping[str, Any] | None = None
+    why_selected: tuple[str, ...] = ()
+    score: float = 0.0
+    char_count: int = 0
+    expandable: bool = False
+    metadata: Mapping[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        if not self.handle.strip() or not self.source_type.strip():
+            raise ValueError("handle and source_type are required")
+        if self.char_count < 0:
+            raise ValueError("char_count cannot be negative")
+        if not math.isfinite(self.score):
+            raise ValueError("score must be finite")
+
+    def to_payload(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass(frozen=True, slots=True)
+class ContextSufficiency:
+    status: ContextStatus
+    missing: tuple[str, ...] = ()
+    conflicts: tuple[str, ...] = ()
+    suggested_queries: tuple[str, ...] = ()
+
+    def to_payload(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass(frozen=True, slots=True)
+class ContextReceipt:
+    revision: str
+    candidate_count: int
+    selected_count: int
+    estimated_tokens: int
+    cache_hit: bool = False
+    semantic_calls: int = 0
+    fallback_reason: str | None = None
+    omitted_count: int = 0
+    detail_reads: int = 0
+
+    def __post_init__(self) -> None:
+        counters = (
+            self.candidate_count,
+            self.selected_count,
+            self.estimated_tokens,
+            self.semantic_calls,
+            self.omitted_count,
+            self.detail_reads,
+        )
+        if any(value < 0 for value in counters):
+            raise ValueError("context receipt counters cannot be negative")
+
+    def to_payload(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass(frozen=True, slots=True)
+class ContextPage:
+    revision: str
+    items: tuple[ContextCard, ...]
+    total: int
+    returned: int
+    has_more: bool
+    next_cursor: str | None = None
+    offset: int = 0
+
+    def __post_init__(self) -> None:
+        if (
+            self.total < 0
+            or self.offset < 0
+            or self.returned != len(self.items)
+            or self.offset + self.returned > self.total
+        ):
+            raise ValueError("invalid context page counts")
+        if self.has_more != (self.offset + self.returned < self.total):
+            raise ValueError("has_more does not match page counts")
+        if self.has_more != (self.next_cursor is not None):
+            raise ValueError("next_cursor must be present exactly when has_more is true")
+
+    def to_payload(self) -> dict[str, Any]:
+        payload = asdict(self)
+        payload["items"] = [item.to_payload() for item in self.items]
+        return payload
+
+
+@dataclass(frozen=True, slots=True)
+class ContextBundle:
+    revision: str
+    cards: tuple[ContextCard, ...]
+    required_handles: tuple[str, ...]
+    sufficiency: ContextSufficiency
+    receipt: ContextReceipt
+    page: ContextPage | None = None
+
+    def to_payload(self) -> dict[str, Any]:
+        return {
+            "revision": self.revision,
+            "cards": [card.to_payload() for card in self.cards],
+            "required_handles": list(self.required_handles),
+            "sufficiency": self.sufficiency.to_payload(),
+            "receipt": self.receipt.to_payload(),
+            "page": self.page.to_payload() if self.page is not None else None,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class ContextCursor:
+    last_sort_key: tuple[str, ...]
+
+
+class ContextCursorError(ValueError):
+    def __init__(self, code: str, message: str) -> None:
+        super().__init__(message)
+        self.code = code
+
+
+class ContextCursorCodec:
+    """Encode continuation state as a signed, caller-opaque token."""
+
+    def __init__(self, secret: bytes) -> None:
+        if len(secret) < 16:
+            raise ValueError("cursor secret must contain at least 16 bytes")
+        self._secret = secret
+
+    def encode(
+        self,
+        *,
+        root_key: str,
+        revision: str,
+        view: str,
+        filter_digest: str,
+        last_sort_key: Sequence[str],
+    ) -> str:
+        payload = _canonical_json(
+            {
+                "v": _CURSOR_VERSION,
+                "r": _binding_digest(root_key),
+                "rev": _binding_digest(revision),
+                "view": view,
+                "filter": filter_digest,
+                "last": list(last_sort_key),
+            }
+        ).encode()
+        signature = hmac.digest(self._secret, payload, "sha256")
+        return f"ctx{_CURSOR_VERSION}.{_b64encode(payload)}.{_b64encode(signature)}"
+
+    def decode(
+        self,
+        token: str,
+        *,
+        expected_root_key: str,
+        expected_revision: str,
+        expected_view: str,
+        expected_filter_digest: str,
+    ) -> ContextCursor:
+        try:
+            prefix, encoded_payload, encoded_signature = token.split(".")
+            if prefix != f"ctx{_CURSOR_VERSION}":
+                raise ValueError("unsupported cursor version")
+            payload = _b64decode(encoded_payload)
+            signature = _b64decode(encoded_signature)
+            if _b64encode(payload) != encoded_payload or _b64encode(signature) != encoded_signature:
+                raise ValueError("non-canonical cursor encoding")
+            if not hmac.compare_digest(signature, hmac.digest(self._secret, payload, "sha256")):
+                raise ValueError("invalid cursor signature")
+            value = json.loads(payload)
+            if not isinstance(value, dict) or value.get("v") != _CURSOR_VERSION:
+                raise ValueError("invalid cursor payload")
+        except (ValueError, TypeError, json.JSONDecodeError) as exc:
+            raise ContextCursorError("CONTEXT_CURSOR_INVALID", "invalid context cursor") from exc
+
+        if value.get("rev") != _binding_digest(expected_revision):
+            raise ContextCursorError(
+                "STALE_WORKBENCH_STATE", "context cursor references an obsolete revision"
+            )
+        if (
+            value.get("r") != _binding_digest(expected_root_key)
+            or value.get("view") != expected_view
+            or value.get("filter") != expected_filter_digest
+        ):
+            raise ContextCursorError(
+                "CONTEXT_CURSOR_SCOPE_MISMATCH",
+                "context cursor does not belong to this root or query",
+            )
+        last = value.get("last")
+        if not isinstance(last, list) or not last:
+            raise ContextCursorError("CONTEXT_CURSOR_INVALID", "invalid cursor sort key")
+        if not all(isinstance(item, str) for item in last):
+            raise ContextCursorError("CONTEXT_CURSOR_INVALID", "invalid cursor sort key")
+        return ContextCursor(last_sort_key=tuple(last))
+
+
+def estimate_context_tokens(value: object, *, safety_factor: float = 1.25) -> int:
+    """Conservatively estimate tokens for mixed Chinese and ASCII JSON content."""
+
+    if safety_factor < 1:
+        raise ValueError("safety_factor cannot be less than 1")
+    text = value if isinstance(value, str) else _canonical_json(value)
+    if not text:
+        return 0
+    ascii_count = sum(ord(char) < 128 for char in text)
+    non_ascii_count = len(text) - ascii_count
+    base = math.ceil(ascii_count / 3) + non_ascii_count
+    return math.ceil(base * safety_factor)
+
+
+def tokenize_context(text: str) -> tuple[str, ...]:
+    """Tokenize ASCII terms plus Chinese unigrams and bigrams without dependencies."""
+
+    tokens = [match.group(0).casefold() for match in _ASCII_TERM.finditer(text)]
+    for match in _CJK_RUN.finditer(text):
+        run = match.group(0)
+        tokens.extend(run)
+        tokens.extend(run[index : index + 2] for index in range(len(run) - 1))
+    return tuple(tokens)
+
+
+def rank_context_cards(
+    cards: Sequence[ContextCard],
+    query: str,
+    *,
+    limit: int | None = None,
+    k1: float = 1.5,
+    b: float = 0.75,
+) -> tuple[ContextCard, ...]:
+    """Rank cards with BM25, exact-content deduplication, and source diversity."""
+
+    if limit is not None and limit < 1:
+        raise ValueError("limit must be positive")
+    unique = _deduplicate_cards(cards)
+    if not unique:
+        return ()
+
+    documents = [tokenize_context(_searchable_text(card)) for card in unique]
+    query_terms = Counter(tokenize_context(query))
+    scores = _bm25_scores(documents, query_terms, k1=k1, b=b)
+    remaining = [
+        (card, lexical_score if query_terms else card.score)
+        for card, lexical_score in zip(unique, scores, strict=True)
+    ]
+    source_counts: Counter[str] = Counter()
+    selected: list[ContextCard] = []
+    selection_limit = min(len(remaining), limit or len(remaining))
+
+    while remaining and len(selected) < selection_limit:
+        remaining.sort(
+            key=lambda item: (
+                -(item[1] / (1 + 0.25 * source_counts[item[0].source_type])),
+                item[0].source_type,
+                item[0].handle,
+            )
+        )
+        card, raw_score = remaining.pop(0)
+        adjusted_score = raw_score / (1 + 0.25 * source_counts[card.source_type])
+        metadata = dict(card.metadata)
+        metadata["lexical_score"] = raw_score
+        selected.append(replace(card, score=adjusted_score, metadata=metadata))
+        source_counts[card.source_type] += 1
+    return tuple(selected)
+
+
+def fit_context_cards(
+    cards: Sequence[ContextCard], token_budget: int
+) -> tuple[tuple[ContextCard, ...], int, int]:
+    """Fit whole cards into a budget; never silently truncate an individual card."""
+
+    if token_budget <= 0:
+        raise ValueError("token_budget must be positive")
+    selected: list[ContextCard] = []
+    used = 0
+    for card in cards:
+        cost = estimate_context_tokens(card.to_payload())
+        if used + cost > token_budget:
+            continue
+        selected.append(card)
+        used += cost
+    return tuple(selected), used, len(cards) - len(selected)
+
+
+def assess_context_sufficiency(
+    *,
+    required_source_types: Sequence[str],
+    cards: Sequence[ContextCard],
+    has_more: bool,
+) -> ContextSufficiency:
+    present = {card.source_type for card in cards}
+    missing = tuple(sorted(set(required_source_types) - present))
+    if not missing:
+        return ContextSufficiency(status="sufficient")
+    if has_more:
+        return ContextSufficiency(
+            status="ambiguous",
+            missing=missing,
+            suggested_queries=tuple(f"source_type:{item}" for item in missing),
+        )
+    return ContextSufficiency(status="insufficient", missing=missing)
+
+
+def paginate_context_cards(
+    cards: Sequence[ContextCard],
+    *,
+    request: ContextRequest,
+    revision: str,
+    cursor_codec: ContextCursorCodec,
+) -> ContextPage:
+    ordered = sorted(cards, key=_card_sort_key)
+    start = 0
+    if request.cursor is not None:
+        decoded = cursor_codec.decode(
+            request.cursor,
+            expected_root_key=request.root_trace_id,
+            expected_revision=revision,
+            expected_view=request.view,
+            expected_filter_digest=request.filter_digest(),
+        )
+        try:
+            start = next(
+                index + 1
+                for index, card in enumerate(ordered)
+                if _serialized_sort_key(card) == decoded.last_sort_key
+            )
+        except StopIteration as exc:
+            raise ContextCursorError(
+                "STALE_WORKBENCH_STATE", "cursor position is no longer present"
+            ) from exc
+
+    selected: list[ContextCard] = []
+    card_budget = max(1, request.token_budget - _PAGE_ENVELOPE_TOKEN_RESERVE)
+    for card in ordered[start : start + request.page_size]:
+        bounded_card = _page_card(card, card_budget) if not selected else card
+        trial = (*selected, bounded_card)
+        if estimate_context_tokens([item.to_payload() for item in trial]) > card_budget:
+            break
+        selected.append(bounded_card)
+    items = tuple(selected)
+    total = len(ordered)
+    has_more = start + len(items) < total
+    next_cursor = None
+    if has_more and items:
+        next_cursor = cursor_codec.encode(
+            root_key=request.root_trace_id,
+            revision=revision,
+            view=request.view,
+            filter_digest=request.filter_digest(),
+            last_sort_key=_serialized_sort_key(items[-1]),
+        )
+    return ContextPage(
+        revision=revision,
+        items=items,
+        total=total,
+        returned=len(items),
+        has_more=has_more,
+        next_cursor=next_cursor,
+        offset=start,
+    )
+
+
+def canonical_digest(value: object) -> str:
+    return "sha256:" + sha256(_canonical_json(value).encode()).hexdigest()
+
+
+def _page_card(card: ContextCard, token_budget: int) -> ContextCard:
+    if estimate_context_tokens(card.to_payload()) <= token_budget:
+        return card
+    compact = replace(
+        card,
+        summary=card.summary[:160],
+        excerpt="",
+        why_selected=(),
+        metadata={"detail_required": True},
+        expandable=True,
+    )
+    if estimate_context_tokens(compact.to_payload()) <= token_budget:
+        return compact
+    minimal = replace(compact, summary=card.summary[:80], goal_ids=(), scope_selector=None)
+    if estimate_context_tokens(minimal.to_payload()) <= token_budget:
+        return minimal
+    raise ContextCursorError(
+        "CONTEXT_BUDGET_EXCEEDED", "context card identity exceeds the page token budget"
+    )
+
+
+def _bm25_scores(
+    documents: Sequence[Sequence[str]],
+    query_terms: Counter[str],
+    *,
+    k1: float,
+    b: float,
+) -> list[float]:
+    if not query_terms:
+        return [0.0] * len(documents)
+    document_count = len(documents)
+    average_length = sum(len(document) for document in documents) / max(document_count, 1)
+    document_frequency = Counter(
+        term for document in documents for term in set(document) if term in query_terms
+    )
+    output: list[float] = []
+    for document in documents:
+        frequencies = Counter(document)
+        score = 0.0
+        for term, query_frequency in query_terms.items():
+            frequency = frequencies[term]
+            if not frequency:
+                continue
+            frequency_count = document_frequency[term]
+            inverse_frequency = math.log(
+                1 + (document_count - frequency_count + 0.5) / (frequency_count + 0.5)
+            )
+            length_ratio = len(document) / average_length if average_length else 0.0
+            denominator = frequency + k1 * (1 - b + b * length_ratio)
+            score += inverse_frequency * (frequency * (k1 + 1) / denominator) * query_frequency
+        output.append(score)
+    return output
+
+
+def _deduplicate_cards(cards: Sequence[ContextCard]) -> list[ContextCard]:
+    by_content: dict[str, ContextCard] = {}
+    for card in cards:
+        normalized = " ".join(f"{card.summary} {card.excerpt}".casefold().split())
+        key = sha256(normalized.encode()).hexdigest() if normalized else card.handle
+        current = by_content.get(key)
+        if current is None or (card.score, card.handle) > (current.score, current.handle):
+            by_content[key] = card
+    return sorted(by_content.values(), key=lambda card: (card.source_type, card.handle))
+
+
+def _searchable_text(card: ContextCard) -> str:
+    metadata = " ".join(
+        str(value) for value in card.metadata.values() if isinstance(value, (str, int, float))
+    )
+    return " ".join((card.source_type, card.summary, card.excerpt, *card.goal_ids, metadata))
+
+
+def _card_sort_key(card: ContextCard) -> tuple[float, str, str]:
+    return (-card.score, card.source_type, card.handle)
+
+
+def _serialized_sort_key(card: ContextCard) -> tuple[str, ...]:
+    return (format(card.score, ".12g"), card.source_type, card.handle)
+
+
+def _canonical_json(value: object) -> str:
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    )
+
+
+def _binding_digest(value: str) -> str:
+    return sha256(value.encode()).hexdigest()
+
+
+def _b64encode(value: bytes) -> str:
+    return base64.urlsafe_b64encode(value).rstrip(b"=").decode()
+
+
+def _b64decode(value: str) -> bytes:
+    return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
+
+
+__all__ = [
+    "ContextBundle",
+    "ContextCard",
+    "ContextCursorCodec",
+    "ContextCursorError",
+    "ContextPage",
+    "ContextReceipt",
+    "ContextRequest",
+    "ContextSufficiency",
+    "assess_context_sufficiency",
+    "canonical_digest",
+    "estimate_context_tokens",
+    "fit_context_cards",
+    "paginate_context_cards",
+    "rank_context_cards",
+    "tokenize_context",
+]

script_build_host/tests/test_context_broker_core.py

@@ -0,0 +1,284 @@
+from __future__ import annotations
+
+from dataclasses import replace
+
+import pytest
+
+from script_build_host.domain.context_broker import (
+    ContextBundle,
+    ContextCard,
+    ContextCursorCodec,
+    ContextCursorError,
+    ContextReceipt,
+    ContextRequest,
+    ContextSufficiency,
+    assess_context_sufficiency,
+    estimate_context_tokens,
+    fit_context_cards,
+    paginate_context_cards,
+    rank_context_cards,
+    tokenize_context,
+)
+
+
+def _card(
+    index: int,
+    *,
+    source_type: str = "pattern",
+    summary: str | None = None,
+    score: float = 0.0,
+) -> ContextCard:
+    return ContextCard(
+        handle=f"source-{index:03d}",
+        source_type=source_type,
+        summary=summary or f"candidate {index}",
+        excerpt=f"detail {index}",
+        goal_ids=("goal-1",),
+        scope_selector={"anchor": "mission", "path": ["opening"]},
+        score=score,
+        char_count=100,
+        expandable=True,
+    )
+
+
+def _request(**changes: object) -> ContextRequest:
+    values: dict[str, object] = {
+        "root_trace_id": "root-49",
+        "role": "planner",
+        "view": "patterns",
+        "query": "AI 叙事",
+        "source_types": ("pattern",),
+        "page_size": 7,
+        "token_budget": 2_000,
+    }
+    values.update(changes)
+    return ContextRequest(**values)  # type: ignore[arg-type]
+
+
+def test_models_validate_limits_and_serialize_without_framework_types() -> None:
+    request = _request(goal_ids=("goal-1",), detail_level="semantic")
+    card = _card(1)
+    sufficiency = ContextSufficiency(status="sufficient")
+    receipt = ContextReceipt(
+        revision="ledger:2",
+        candidate_count=1,
+        selected_count=1,
+        estimated_tokens=30,
+    )
+    bundle = ContextBundle(
+        revision="ledger:2",
+        cards=(card,),
+        required_handles=(card.handle,),
+        sufficiency=sufficiency,
+        receipt=receipt,
+    )
+
+    assert request.filter_digest().startswith("sha256:")
+    assert bundle.to_payload()["cards"][0]["handle"] == "source-001"
+    assert bundle.to_payload()["receipt"]["semantic_calls"] == 0
+    with pytest.raises(ValueError, match="page_size"):
+        _request(page_size=21)
+    with pytest.raises(ValueError, match="unique"):
+        _request(goal_ids=("g", "g"))
+
+
+def test_token_estimate_is_conservative_for_chinese_and_ascii() -> None:
+    chinese = "中文创作需要完整上下文"
+    ascii_text = "a" * 300
+
+    assert estimate_context_tokens(chinese) >= len(chinese)
+    assert estimate_context_tokens(ascii_text) >= 125
+    assert estimate_context_tokens("") == 0
+    with pytest.raises(ValueError, match="safety_factor"):
+        estimate_context_tokens("text", safety_factor=0.9)
+
+
+def test_tokenizer_and_bm25_rank_mixed_chinese_ascii_queries() -> None:
+    tokens = tokenize_context("AI叙事结构")
+    assert {"ai", "叙", "事", "叙事", "结构"}.issubset(tokens)
+    cards = (
+        _card(1, summary="AI 叙事结构和开场钩子"),
+        _card(2, summary="健身饮食与休息"),
+        _card(3, summary="叙事节奏和故事结构"),
+    )
+
+    ranked = rank_context_cards(cards, "AI 叙事结构")
+
+    assert ranked[0].handle == "source-001"
+    assert ranked[0].score > ranked[-1].score
+    assert ranked[0].metadata["lexical_score"] > 0
+
+
+def test_ranking_deduplicates_content_and_penalizes_one_dominant_source() -> None:
+    cards = (
+        _card(1, source_type="pattern", summary="创作脚本", score=10),
+        replace(
+            _card(2, source_type="pattern", summary="创作脚本", score=9),
+            excerpt="detail 1",
+        ),
+        _card(3, source_type="pattern", summary="创作开场", score=8),
+        _card(4, source_type="evidence", summary="创作证据", score=7.9),
+    )
+
+    ranked = rank_context_cards(cards, "", limit=3)
+
+    assert {card.handle for card in ranked}.isdisjoint({"source-002"})
+    assert [card.source_type for card in ranked[:2]] == ["pattern", "evidence"]
+
+
+def test_budget_keeps_cards_whole_and_reports_omissions() -> None:
+    cards = tuple(_card(index, summary="正文" * 100) for index in range(3))
+    one_card_budget = estimate_context_tokens(cards[0].to_payload()) + 5
+
+    selected, used, omitted = fit_context_cards(cards, one_card_budget)
+
+    assert len(selected) == 1
+    assert used <= one_card_budget
+    assert omitted == 2
+
+
+def test_pagination_shrinks_page_to_token_budget_without_losing_cards() -> None:
+    cards = tuple(_card(index, summary="详细中文上下文" * 80) for index in range(20))
+    codec = ContextCursorCodec(b"page-budget-secret")
+    request = _request(page_size=20, token_budget=2_000)
+    seen: list[str] = []
+    cursor = None
+
+    while True:
+        page = paginate_context_cards(
+            cards,
+            request=replace(request, cursor=cursor),
+            revision="ledger:1",
+            cursor_codec=codec,
+        )
+        assert page.items
+        assert estimate_context_tokens([item.to_payload() for item in page.items]) <= 2_000
+        seen.extend(item.handle for item in page.items)
+        cursor = page.next_cursor
+        if cursor is None:
+            break
+
+    assert seen == sorted(seen)
+    assert len(seen) == len(set(seen)) == 20
+
+
+def test_oversized_first_card_degrades_to_expandable_outline() -> None:
+    card = replace(
+        _card(1),
+        summary="摘要" * 600,
+        excerpt="正文" * 1_500,
+        metadata={"large": "元数据" * 300},
+    )
+
+    page = paginate_context_cards(
+        (card,),
+        request=_request(page_size=1, token_budget=2_000),
+        revision="artifact:1",
+        cursor_codec=ContextCursorCodec(b"oversized-card-secret"),
+    )
+
+    assert page.returned == 1
+    assert page.items[0].handle == card.handle
+    assert page.items[0].expandable is True
+    assert page.items[0].metadata == {"detail_required": True}
+    assert estimate_context_tokens(page.items[0].to_payload()) <= 1_500
+
+
+def test_sufficiency_distinguishes_expandable_and_exhausted_missing_sources() -> None:
+    cards = (_card(1, source_type="direction"),)
+
+    ambiguous = assess_context_sufficiency(
+        required_source_types=("direction", "evidence"), cards=cards, has_more=True
+    )
+    insufficient = assess_context_sufficiency(
+        required_source_types=("direction", "evidence"), cards=cards, has_more=False
+    )
+    sufficient = assess_context_sufficiency(
+        required_source_types=("direction",), cards=cards, has_more=False
+    )
+
+    assert ambiguous.status == "ambiguous"
+    assert ambiguous.suggested_queries == ("source_type:evidence",)
+    assert insufficient.status == "insufficient"
+    assert sufficient.status == "sufficient"
+
+
+def test_signed_cursor_traverses_every_item_without_omission_or_duplicate() -> None:
+    codec = ContextCursorCodec(b"0123456789abcdef-context")
+    cards = tuple(replace(_card(index), score=float(40 - index)) for index in range(33))
+    request = _request()
+    handles: list[str] = []
+    offsets: list[int] = []
+
+    while True:
+        page = paginate_context_cards(
+            cards, request=request, revision="ledger:17", cursor_codec=codec
+        )
+        handles.extend(card.handle for card in page.items)
+        offsets.append(page.offset)
+        if not page.has_more:
+            break
+        assert page.next_cursor is not None
+        request = replace(request, cursor=page.next_cursor)
+
+    assert len(handles) == 33
+    assert len(set(handles)) == 33
+    assert offsets == [0, 7, 14, 21, 28]
+
+
+@pytest.mark.parametrize(
+    ("change", "code"),
+    [
+        ({"root_trace_id": "other-root"}, "CONTEXT_CURSOR_SCOPE_MISMATCH"),
+        ({"view": "tasks"}, "CONTEXT_CURSOR_SCOPE_MISMATCH"),
+        ({"query": "different"}, "CONTEXT_CURSOR_SCOPE_MISMATCH"),
+    ],
+)
+def test_cursor_is_bound_to_root_view_and_filters(change: dict[str, object], code: str) -> None:
+    codec = ContextCursorCodec(b"0123456789abcdef-context")
+    request = _request(page_size=1)
+    first = paginate_context_cards(
+        (_card(1, score=2), _card(2, score=1)),
+        request=request,
+        revision="ledger:17",
+        cursor_codec=codec,
+    )
+    assert first.next_cursor is not None
+    changed = replace(request, cursor=first.next_cursor, **change)
+
+    with pytest.raises(ContextCursorError) as caught:
+        paginate_context_cards(
+            (_card(1, score=2), _card(2, score=1)),
+            request=changed,
+            revision="ledger:17",
+            cursor_codec=codec,
+        )
+    assert caught.value.code == code
+
+
+def test_cursor_rejects_stale_revision_and_tampering() -> None:
+    codec = ContextCursorCodec(b"0123456789abcdef-context")
+    cards = (_card(1, score=2), _card(2, score=1))
+    request = _request(page_size=1)
+    first = paginate_context_cards(cards, request=request, revision="ledger:17", cursor_codec=codec)
+    assert first.next_cursor is not None
+
+    with pytest.raises(ContextCursorError) as stale:
+        paginate_context_cards(
+            cards,
+            request=replace(request, cursor=first.next_cursor),
+            revision="ledger:18",
+            cursor_codec=codec,
+        )
+    assert stale.value.code == "STALE_WORKBENCH_STATE"
+
+    replacement = "A" if first.next_cursor[-1] != "A" else "B"
+    tampered = first.next_cursor[:-1] + replacement
+    with pytest.raises(ContextCursorError) as invalid:
+        paginate_context_cards(
+            cards,
+            request=replace(request, cursor=tampered),
+            revision="ledger:17",
+            cursor_codec=codec,
+        )
+    assert invalid.value.code == "CONTEXT_CURSOR_INVALID"