feat(acp): 模型驱动上下文压缩机制（双机制并存 + 可恢复）

backend/.env.example

@@ -28,3 +28,6 @@ WECHAT_ILINK_BASE_URL="https://ilinkai.weixin.qq.com"
 CHANNEL_DELIVERY_POLL_SECONDS="1.0"
 CHANNEL_DELIVERY_MAX_ATTEMPTS="8"
 CHANNEL_RICH_RENDER_ENABLED="true"
+# ACP（Active Context Pruning）上下文压缩机制全局开关。默认关闭：偏好被忽略、
+# 强制走 legacy 固定阈值压缩；开启后运行设置页可选 ACP 并暴露阈值配置。
+ACP_ENABLED="false"

backend/app/api/ui_config.py

@@ -12,6 +12,7 @@
 from sqlmodel import Session
 
 from app import paths
+from app.config import get_settings
 from app.db import get_session
 from app.db.models import UIConfig, User, utc_now
 from app.harness.sandbox import diagnostics, windows_install_command
@@ -55,6 +56,12 @@ class UIConfigRead(BaseModel):
     sandbox_status_code: str | None = None
     sandbox_status_message: str | None = None
     sandbox_status_remediation: str | None = None
+    context_compression_mode: Literal["acp", "legacy"] = "legacy"
+    acp_model_context_limit: int = 128000
+    acp_nudge_max_pct: float = 0.70
+    acp_nudge_emergency_pct: float = 0.85
+    acp_nudge_min_pct: float = 0.45
+    acp_enabled: bool = False
     updated_at: str
 
     model_config = ConfigDict(from_attributes=True)
@@ -91,6 +98,23 @@ class UIConfigUpdateRequest(BaseModel):
     harness_storage_path: str = Field(default="", max_length=1024)
     sandbox_network_mode: Literal["all", "allowlist", "deny"] = "all"
     sandbox_allowed_domains: list[str] = Field(default_factory=list, max_length=200)
+    context_compression_mode: Literal["acp", "legacy"] = "legacy"
+    acp_model_context_limit: int = Field(default=128000, ge=1)
+    acp_nudge_max_pct: float = Field(default=0.70, ge=0.0, le=1.0)
+    acp_nudge_emergency_pct: float = Field(default=0.85, ge=0.0, le=1.0)
+    acp_nudge_min_pct: float = Field(default=0.45, ge=0.0, le=1.0)
+
+    @model_validator(mode="after")
+    def _validate_acp_threshold_order(self) -> "UIConfigUpdateRequest":
+        if not (
+            0.0
+            <= self.acp_nudge_min_pct
+            <= self.acp_nudge_max_pct
+            <= self.acp_nudge_emergency_pct
+            <= 1.0
+        ):
+            raise ValueError("ACP 阈值必须满足 0 ≤ min ≤ max ≤ emergency ≤ 1")
+        return self
 
     @model_validator(mode="after")
     def validate_context_budgets(self) -> UIConfigUpdateRequest:
@@ -176,6 +200,16 @@ def ui_config_read(row: UIConfig, *, restart_scheduled: bool = False) -> UIConfi
             report.message if report is not None else "沙盒已由管理员关闭。"
         ),
         sandbox_status_remediation=report.remediation if report is not None else None,
+        context_compression_mode=(
+            row.context_compression_mode
+            if row.context_compression_mode in {"acp", "legacy"}
+            else "legacy"
+        ),
+        acp_model_context_limit=row.acp_model_context_limit,
+        acp_nudge_max_pct=row.acp_nudge_max_pct,
+        acp_nudge_emergency_pct=row.acp_nudge_emergency_pct,
+        acp_nudge_min_pct=row.acp_nudge_min_pct,
+        acp_enabled=get_settings().acp_enabled,
         updated_at=row.updated_at.isoformat(),
     )
 
@@ -225,6 +259,11 @@ def update_enterprise_ui_config(
     row.harness_storage_path = storage_path
     row.sandbox_network_mode = request.sandbox_network_mode
     row.sandbox_allowed_domains = request.sandbox_allowed_domains
+    row.context_compression_mode = request.context_compression_mode
+    row.acp_model_context_limit = request.acp_model_context_limit
+    row.acp_nudge_max_pct = request.acp_nudge_max_pct
+    row.acp_nudge_emergency_pct = request.acp_nudge_emergency_pct
+    row.acp_nudge_min_pct = request.acp_nudge_min_pct
     row.updated_at = utc_now()
     db.add(row)
     db.commit()

backend/app/config.py

@@ -61,6 +61,9 @@ class Settings(BaseSettings):
     # 回滚为逐行展示的旧样式；binding 的 config_json.compact_trace=false 可对单个
     # 绑定回滚。
     channel_feishu_trace_compact_sop: bool = True
+    # ACP（Active Context Pruning）上下文压缩机制全局开关：关闭时偏好被忽略、
+    # 强制走 legacy 固定阈值压缩；状态经 ui-config API 暴露给前端置灰 ACP 选项。
+    acp_enabled: bool = False
 
     model_config = SettingsConfigDict(
         env_file=_os.environ.get("ULTRARAG_DOTENV", ".env"),

backend/app/core/acp/__init__.py

@@ -0,0 +1,43 @@
+"""ACP (Active Context Pruning) framework-agnostic compression kernel.
+
+The kernel is standalone: it depends only on the Python standard library
+and never imports staffDeck business modules. The model (caller) writes
+summaries; the kernel handles block management, checkpoints, retrieval,
+nudging, and shadow-price accounting.
+"""
+
+from .blocks import Block, BlockStore
+from .checkpoint import CheckpointRecord, CheckpointStore
+from .config import AcpConfig
+from .engine import AcpEngine, AcpError, CompressResult, DecompressResult, SearchResult
+from .nudge import NudgeRecommendation
+from .pricing import AcpLedger, LedgerEntry, TokenMeter
+from .search import SearchHit, search_blocks
+from .status import BlockLedgerEntry, CheckpointMappingEntry, ContextStatus
+from .tiers import MAX_TIER, can_distill, next_tier, tier_label
+
+__all__ = [
+    "AcpConfig",
+    "AcpEngine",
+    "AcpError",
+    "AcpLedger",
+    "Block",
+    "BlockLedgerEntry",
+    "BlockStore",
+    "CheckpointMappingEntry",
+    "CheckpointRecord",
+    "CheckpointStore",
+    "CompressResult",
+    "ContextStatus",
+    "DecompressResult",
+    "LedgerEntry",
+    "MAX_TIER",
+    "NudgeRecommendation",
+    "SearchHit",
+    "SearchResult",
+    "TokenMeter",
+    "can_distill",
+    "next_tier",
+    "search_blocks",
+    "tier_label",
+]

backend/app/core/acp/blocks.py

@@ -0,0 +1,80 @@
+"""Message-level atomic blocks and the ordered block store.
+
+Each message added to the kernel becomes exactly one block; blocks are the
+atomic unit of compression. staffDeck has no standard function calling, so
+block boundaries are message-level rather than tool-call/result pairs.
+"""
+
+from dataclasses import dataclass
+from typing import Sequence
+
+
+@dataclass(frozen=True)
+class Block:
+    """An immutable message-level block.
+
+    ``is_summary`` marks blocks produced by compression; their original
+    content lives in the checkpoint record (``checkpoint_id``) so the
+    visible state stays small while recovery stays possible.
+    """
+
+    block_id: int
+    message_id: str
+    content: str
+    tier: int = 1
+    is_summary: bool = False
+    checkpoint_id: int | None = None
+    skip: bool = False
+
+
+class BlockStore:
+    """Ordered collection of blocks with stable, monotonically increasing ids."""
+
+    def __init__(self) -> None:
+        self._blocks: list[Block] = []
+        self._next_id = 1
+
+    def add(self, message_id: str, content: str, *, skip: bool = False) -> Block:
+        block = Block(
+            block_id=self._next_id,
+            message_id=message_id,
+            content=content,
+            skip=skip,
+        )
+        self._next_id += 1
+        self._blocks.append(block)
+        return block
+
+    def add_many(self, messages: Sequence[tuple[str, str]]) -> list[Block]:
+        return [self.add(message_id, content) for message_id, content in messages]
+
+    def get(self, block_id: int) -> Block | None:
+        for block in self._blocks:
+            if block.block_id == block_id:
+                return block
+        return None
+
+    def all(self) -> tuple[Block, ...]:
+        return tuple(self._blocks)
+
+    def slice(self, seq_start: int, seq_end: int) -> list[Block]:
+        return self._blocks[seq_start : seq_end + 1]
+
+    def replace(self, seq_start: int, seq_end: int, new_block: Block) -> list[Block]:
+        removed = self._blocks[seq_start : seq_end + 1]
+        self._blocks[seq_start : seq_end + 1] = [new_block]
+        return removed
+
+    def replace_by_id(self, block_id: int, replacement: Sequence[Block]) -> Block | None:
+        for index, block in enumerate(self._blocks):
+            if block.block_id == block_id:
+                removed = self._blocks[index]
+                self._blocks[index : index + 1] = list(replacement)
+                return removed
+        return None
+
+    def next_id(self) -> int:
+        return self._next_id
+
+    def __len__(self) -> int:
+        return len(self._blocks)

backend/app/core/acp/checkpoint.py

@@ -0,0 +1,57 @@
+"""Checkpoint records enabling recovery of compressed content.
+
+Every compress operation writes one checkpoint holding the sequence
+mapping (compressed range), the summary block id, and the original blocks
+with their full content. Checkpoints are immutable and never deleted, so
+decompression and audit trails stay possible at any depth.
+"""
+
+import time
+from dataclasses import dataclass, field
+
+from .blocks import Block
+
+
+@dataclass(frozen=True)
+class CheckpointRecord:
+    """Immutable record of one compress operation."""
+
+    checkpoint_id: int
+    seq_start: int
+    seq_end: int
+    summary_block_id: int
+    tier: int
+    original_blocks: tuple[Block, ...]
+    token_delta: int
+    created_at: float = field(default_factory=time.time)
+
+
+class CheckpointStore:
+    """Append-only store of checkpoint records."""
+
+    def __init__(self) -> None:
+        self._records: dict[int, CheckpointRecord] = {}
+        self._next_id = 1
+
+    def add(self, record: CheckpointRecord) -> CheckpointRecord:
+        self._records[record.checkpoint_id] = record
+        self._next_id = max(self._next_id, record.checkpoint_id + 1)
+        return record
+
+    def get(self, checkpoint_id: int) -> CheckpointRecord | None:
+        return self._records.get(checkpoint_id)
+
+    def find_by_summary_block(self, block_id: int) -> CheckpointRecord | None:
+        for record in self._records.values():
+            if record.summary_block_id == block_id:
+                return record
+        return None
+
+    def all(self) -> tuple[CheckpointRecord, ...]:
+        return tuple(self._records.values())
+
+    def next_id(self) -> int:
+        return self._next_id
+
+    def __len__(self) -> int:
+        return len(self._records)

backend/app/core/acp/config.py

@@ -0,0 +1,49 @@
+"""ACP kernel configuration.
+
+Holds the tunable thresholds for context-pressure nudging and lexical
+search. Values are validated eagerly so misconfiguration fails fast at
+construction time instead of surfacing as surprising runtime behaviour.
+"""
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class AcpConfig:
+    """Tunable parameters for the ACP compression kernel.
+
+    Percentages are fractions in ``(0, 1]`` and must satisfy
+    ``min <= max <= emergency`` so the pressure zones stay well ordered.
+    """
+
+    model_context_limit: int = 128000
+    nudge_max_context_limit_pct: float = 0.70
+    nudge_emergency_threshold_pct: float = 0.85
+    nudge_min_context_limit_pct: float = 0.45
+    search_top_k: int = 5
+    search_min_score: float = 0.0
+    search_ngram_size: int = 3
+    max_tier: int = 3
+
+    def __post_init__(self) -> None:
+        if self.model_context_limit <= 0:
+            raise ValueError("model_context_limit must be positive")
+        for name, value in (
+            ("nudge_max_context_limit_pct", self.nudge_max_context_limit_pct),
+            ("nudge_emergency_threshold_pct", self.nudge_emergency_threshold_pct),
+            ("nudge_min_context_limit_pct", self.nudge_min_context_limit_pct),
+        ):
+            if not 0 < value <= 1:
+                raise ValueError(f"{name} must be in (0, 1], got {value}")
+        if not (
+            self.nudge_min_context_limit_pct
+            <= self.nudge_max_context_limit_pct
+            <= self.nudge_emergency_threshold_pct
+        ):
+            raise ValueError("nudge thresholds must satisfy min <= max <= emergency")
+        if self.search_top_k <= 0:
+            raise ValueError("search_top_k must be positive")
+        if self.search_ngram_size < 2:
+            raise ValueError("search_ngram_size must be at least 2")
+        if self.max_tier < 2:
+            raise ValueError("max_tier must be at least 2")