feat(memory): implement dual-mode archival in memory consolidation

CLAUDE.md

@@ -124,7 +124,7 @@ src/synthorg/
   core/           # Shared domain models, base classes, and resilience config (RetryConfig, RateLimiterConfig)
   engine/         # Agent orchestration, execution loops, parallel execution, task decomposition, routing, task assignment, centralized single-writer task state engine (TaskEngine), task lifecycle, recovery, shutdown, workspace isolation, coordination (multi-agent pipeline: TopologyDispatcher protocol, 4 dispatchers — SAS/centralized/decentralized/context-dependent, wave execution, workspace lifecycle integration, CoordinationSectionConfig company config bridge, build_coordinator factory), coordination error classification, prompt policy validation, checkpoint recovery (checkpoint/, per-turn persistence, heartbeat detection, CheckpointRecoveryStrategy), approval gate (escalation detection, context parking/resume, EscalationInfo/ResumePayload models), stagnation detection (stagnation/, StagnationDetector protocol, ToolRepetitionDetector, dual-signal analysis, corrective prompt injection), agent runtime state (AgentRuntimeState, lightweight per-agent execution status for dashboard queries and recovery), context budget management (context_budget.py, ContextBudgetIndicator, fill estimation, token estimation protocol in token_estimation.py), conversation compaction (compaction/, CompactionCallback type alias, CompactionConfig, CompressionMetadata, oldest-turns summarizer)
   hr/             # HR engine: hiring, firing, onboarding, offboarding, agent registry, performance tracking (task metrics, collaboration scoring, LLM calibration sampling, collaboration overrides, trend detection), promotion/demotion (criteria evaluation, approval strategies, model mapping)
-  memory/         # Persistent agent memory (pluggable MemoryBackend protocol), backends/ (Mem0 adapter: backends/mem0/), retrieval pipeline (ranking, RRF fusion, injection, context formatting, non-inferable filtering), shared org memory (org/), consolidation/archival (consolidation/)
+  memory/         # Persistent agent memory (pluggable MemoryBackend protocol), backends/ (Mem0 adapter: backends/mem0/), retrieval pipeline (ranking, RRF fusion, injection, context formatting, non-inferable filtering), shared org memory (org/), consolidation/archival (consolidation/, dual-mode density-aware archival: DensityClassifier, AbstractiveSummarizer, ExtractivePreserver, DualModeConsolidationStrategy)
   persistence/    # Operational data persistence — pluggable PersistenceBackend protocol, SQLite initial, SettingsRepository (namespaced settings CRUD) (see Memory & Persistence design page)
   observability/  # Structured logging, correlation tracking, log sinks
   providers/      # LLM provider abstraction (LiteLLM adapter)

docs/api/memory.md

@@ -76,3 +76,11 @@ Persistent agent memory — protocol, retrieval pipeline, shared org memory, con
 ::: synthorg.memory.consolidation.strategy
 
 ::: synthorg.memory.consolidation.service
+
+::: synthorg.memory.consolidation.density
+
+::: synthorg.memory.consolidation.extractive
+
+::: synthorg.memory.consolidation.abstractive
+
+::: synthorg.memory.consolidation.dual_mode_strategy

docs/architecture/tech-stack.md

@@ -118,7 +118,7 @@ These conventions are used throughout the codebase. For full details on each, se
 | **LLM call analytics** | Adopted | Proxy metrics (`turns_per_task`, `tokens_per_task`) and data models for call categorization, coordination metrics, and orchestration ratio. |
 | **Cost tiers and quota tracking** | Adopted | Configurable `CostTierDefinition` with merge/override semantics. `QuotaTracker` enforces per-provider request/token quotas with window-based rotation. |
 | **Shared org memory** | Adopted | `OrgMemoryBackend` protocol with `HybridPromptRetrievalBackend`. Seniority-based write access control. Core policies in system prompts; extended facts retrieved on demand. |
-| **Memory consolidation** | Adopted | `ConsolidationStrategy` protocol with deduplication + summarization. `RetentionEnforcer` for age-based cleanup. `ArchivalStore` for cold storage. |
+| **Memory consolidation** | Adopted | `ConsolidationStrategy` protocol with simple (deduplication + summarization) and dual-mode (density-aware: abstractive LLM summary for sparse content, extractive preservation for dense content) strategies. `RetentionEnforcer` for age-based cleanup. `ArchivalStore` for cold storage with deterministic index-based restore. |
 | **State coordination** | Adopted | Centralized single-writer `TaskEngine` with `asyncio.Queue`. Agents submit requests; engine applies `model_validate` / `with_transition` sequentially and publishes snapshots. |
 | **Workspace isolation** | Adopted | Pluggable `WorkspaceIsolationStrategy` protocol. Default: git worktrees with sequential merge on completion. |
 | **Graceful shutdown** | Adopted | Pluggable `ShutdownStrategy` protocol with cooperative 30-second timeout. Force-cancel after timeout with `INTERRUPTED` status. |

docs/design/memory.md

@@ -323,7 +323,36 @@ models in `memory/consolidation/config.py`:
 |--------|---------|
 | `ConsolidationConfig` | Top-level: `max_memories_per_agent` limit, nested `retention` and `archival` sub-configs |
 | `RetentionConfig` | Per-category `RetentionRule` tuples (category + retention_days), optional `default_retention_days` fallback |
-| `ArchivalConfig` | Enables/disables archival of consolidated entries to `ArchivalStore` |
+| `ArchivalConfig` | Enables/disables archival of consolidated entries to `ArchivalStore`, nested `DualModeConfig` |
+| `DualModeConfig` | Density-aware dual-mode archival: threshold, summarization model, anchor/fact limits |
+
+#### Dual-Mode Archival
+
+When `ArchivalConfig.dual_mode.enabled` is `True`, consolidation classifies content density before
+choosing an archival mode. This prevents catastrophic information loss from naively summarizing
+dense content (code, structured data, identifiers). Based on research: Memex
+([arXiv:2603.04257](https://arxiv.org/abs/2603.04257)) and KV Cache Attention Matching
+([arXiv:2602.16284](https://arxiv.org/abs/2602.16284)).
+
+| Density | Archival Mode | Method |
+|---------|--------------|--------|
+| Sparse (conversational, narrative) | `ABSTRACTIVE` | LLM-generated summary via `AbstractiveSummarizer` |
+| Dense (code, structured data, IDs) | `EXTRACTIVE` | Verbatim key-fact extraction + start/mid/end anchors via `ExtractivePreserver` |
+
+**Classification** is heuristic-based (`DensityClassifier`), using five weighted signals: code
+patterns, structured data markers, identifier density, numeric density, and line structure.  No LLM
+is needed for classification — only for abstractive summarization.  Groups are classified by
+majority vote: if most entries in a category group are dense, the group uses extractive mode.
+
+**Deterministic restore**: When entries are archived, the service builds an `archival_index`
+(mapping `original_id` → `archival_id`) on `ConsolidationResult`.  Agents can use this index to
+call `ArchivalStore.restore(agent_id, entry_id)` directly by ID, bypassing semantic search.
+
+| Model | Purpose |
+|-------|---------|
+| `ArchivalMode` | Enum: `ABSTRACTIVE` or `EXTRACTIVE` |
+| `ArchivalModeAssignment` | Maps a removed entry ID to its archival mode (set by strategy) |
+| `ArchivalIndexEntry` | Maps original entry ID to archival store ID (built by service) |
 
 !!! abstract "Scope Note"
 

src/synthorg/memory/__init__.py

@@ -20,10 +20,13 @@
     MemoryStorageConfig,
 )
 from synthorg.memory.consolidation import (
+    ArchivalMode,
     ArchivalStore,
     ConsolidationConfig,
     ConsolidationResult,
     ConsolidationStrategy,
+    ContentDensity,
+    DualModeConsolidationStrategy,
     MemoryConsolidationService,
     RetentionEnforcer,
     SimpleConsolidationStrategy,
@@ -70,13 +73,16 @@
 from synthorg.memory.shared import SharedKnowledgeStore
 
 __all__ = [
+    "ArchivalMode",
     "ArchivalStore",
     "CompanyMemoryConfig",
     "ConsolidationConfig",
     "ConsolidationResult",
     "ConsolidationStrategy",
+    "ContentDensity",
     "ContextInjectionStrategy",
     "DefaultTokenEstimator",
+    "DualModeConsolidationStrategy",
     "FusionStrategy",
     "InjectionPoint",
     "InjectionStrategy",

src/synthorg/memory/consolidation/__init__.py

@@ -4,14 +4,24 @@
 ``synthorg.memory.consolidation`` directly.
 """
 
+from synthorg.memory.consolidation.abstractive import AbstractiveSummarizer
 from synthorg.memory.consolidation.archival import ArchivalStore
 from synthorg.memory.consolidation.config import (
     ArchivalConfig,
     ConsolidationConfig,
+    DualModeConfig,
     RetentionConfig,
 )
+from synthorg.memory.consolidation.density import ContentDensity, DensityClassifier
+from synthorg.memory.consolidation.dual_mode_strategy import (
+    DualModeConsolidationStrategy,
+)
+from synthorg.memory.consolidation.extractive import ExtractivePreserver
 from synthorg.memory.consolidation.models import (
     ArchivalEntry,
+    ArchivalIndexEntry,
+    ArchivalMode,
+    ArchivalModeAssignment,
     ConsolidationResult,
     RetentionRule,
 )
@@ -23,12 +33,21 @@
 from synthorg.memory.consolidation.strategy import ConsolidationStrategy
 
 __all__ = [
+    "AbstractiveSummarizer",
     "ArchivalConfig",
     "ArchivalEntry",
+    "ArchivalIndexEntry",
+    "ArchivalMode",
+    "ArchivalModeAssignment",
     "ArchivalStore",
     "ConsolidationConfig",
     "ConsolidationResult",
     "ConsolidationStrategy",
+    "ContentDensity",
+    "DensityClassifier",
+    "DualModeConfig",
+    "DualModeConsolidationStrategy",
+    "ExtractivePreserver",
     "MemoryConsolidationService",
     "RetentionConfig",
     "RetentionEnforcer",

src/synthorg/memory/consolidation/abstractive.py

@@ -0,0 +1,139 @@
+"""Abstractive summarizer for sparse memory content.
+
+Uses an LLM (via ``CompletionProvider``) to generate concise summaries
+of conversational/narrative memory content.  Falls back to truncation
+if the LLM call fails.
+"""
+
+import builtins
+
+from synthorg.core.types import NotBlankStr  # noqa: TC001
+from synthorg.memory.models import MemoryEntry  # noqa: TC001
+from synthorg.observability import get_logger
+from synthorg.observability.events.consolidation import (
+    DUAL_MODE_ABSTRACTIVE_FALLBACK,
+    DUAL_MODE_ABSTRACTIVE_SUMMARY,
+)
+from synthorg.providers.enums import MessageRole
+from synthorg.providers.models import ChatMessage, CompletionConfig
+from synthorg.providers.protocol import CompletionProvider  # noqa: TC001
+
+logger = get_logger(__name__)
+
+_TRUNCATE_LENGTH = 200
+
+_SYSTEM_PROMPT = (
+    "You are a memory consolidation assistant. Summarize the following "
+    "memory content concisely, preserving key decisions, events, and "
+    "learnings. Be factual, specific, and brief."
+)
+
+
+def _truncate_fallback(content: str) -> str:
+    """Truncate content as a fallback when LLM summarization fails."""
+    if len(content) <= _TRUNCATE_LENGTH:
+        return content
+    return content[:_TRUNCATE_LENGTH] + "..."
+
+
+class AbstractiveSummarizer:
+    """LLM-based abstractive summarizer for sparse content.
+
+    Uses a ``CompletionProvider`` to generate concise summaries of
+    conversational/narrative memory content.  Falls back to truncation
+    if the LLM call fails.
+
+    Args:
+        provider: Completion provider for LLM calls.
+        model: Model identifier to use for summarization.
+        max_summary_tokens: Maximum tokens for the summary response.
+        temperature: Sampling temperature for summarization.
+    """
+
+    def __init__(
+        self,
+        *,
+        provider: CompletionProvider,
+        model: str,
+        max_summary_tokens: int = 200,
+        temperature: float = 0.3,
+    ) -> None:
+        if not model or not model.strip():
+            msg = "model must be a non-blank string"
+            raise ValueError(msg)
+        self._provider = provider
+        self._model = model
+        self._config = CompletionConfig(
+            temperature=temperature,
+            max_tokens=max_summary_tokens,
+        )
+
+    async def summarize(self, content: str) -> str:
+        """Generate an abstractive summary of the given content.
+
+        Falls back to truncation if the LLM call fails or returns
+        empty content.
+
+        Args:
+            content: The sparse/conversational text to summarize.
+
+        Returns:
+            Summary text.
+        """
+        try:
+            messages = [
+                ChatMessage(role=MessageRole.SYSTEM, content=_SYSTEM_PROMPT),
+                ChatMessage(role=MessageRole.USER, content=content),
+            ]
+            response = await self._provider.complete(
+                messages,
+                self._model,
+                config=self._config,
+            )
+            if response.content and response.content.strip():
+                logger.debug(
+                    DUAL_MODE_ABSTRACTIVE_SUMMARY,
+                    content_length=len(content),
+                    summary_length=len(response.content),
+                    model=self._model,
+                )
+                return response.content.strip()
+        except builtins.MemoryError, RecursionError:
+            raise
+        except Exception as exc:
+            logger.warning(
+                DUAL_MODE_ABSTRACTIVE_FALLBACK,
+                content_length=len(content),
+                error=str(exc),
+                error_type=type(exc).__name__,
+            )
+
+        # Fallback: truncation
+        logger.debug(
+            DUAL_MODE_ABSTRACTIVE_FALLBACK,
+            content_length=len(content),
+            reason="empty_or_failed",
+        )
+        return _truncate_fallback(content)
+
+    async def summarize_batch(
+        self,
+        entries: tuple[MemoryEntry, ...],
+    ) -> tuple[tuple[NotBlankStr, str], ...]:
+        """Summarize multiple entries.
+
+        Each entry is summarized independently.  Failures for
+        individual entries fall back to truncation without aborting
+        the batch.
+
+        Args:
+            entries: Memory entries to summarize.
+
+        Returns:
+            Tuple of ``(entry_id, summary)`` pairs in input order.
+        """
+        results: list[tuple[NotBlankStr, str]] = []
+        for entry in entries:
+            summary = await self.summarize(entry.content)
+            results.append((entry.id, summary))
+        return tuple(results)

src/synthorg/memory/consolidation/config.py

@@ -48,12 +48,77 @@ def _validate_unique_categories(self) -> Self:
         return self
 
 
+class DualModeConfig(BaseModel):
+    """Configuration for dual-mode archival.
+
+    Controls density-aware archival: LLM abstractive summaries for
+    sparse/conversational content vs extractive preservation (verbatim
+    key facts + start/mid/end anchors) for dense/factual content.
+
+    Attributes:
+        enabled: Whether dual-mode density classification is active.
+            When ``False``, all entries use abstractive mode.
+        dense_threshold: Density score threshold for DENSE classification
+            (0.0 = classify everything as dense, 1.0 = everything sparse).
+        summarization_model: Model ID for abstractive summarization.
+        max_summary_tokens: Maximum tokens for LLM summary responses.
+        max_facts: Maximum number of extracted key facts for extractive
+            mode.
+        anchor_length: Character length for each extractive anchor
+            snippet (start/mid/end).
+    """
+
+    model_config = ConfigDict(frozen=True, allow_inf_nan=False)
+
+    enabled: bool = Field(
+        default=False,
+        description="Whether dual-mode density classification is active",
+    )
+    dense_threshold: float = Field(
+        default=0.5,
+        ge=0.0,
+        le=1.0,
+        description="Density score threshold for DENSE classification",
+    )
+    summarization_model: str = Field(
+        default="",
+        description="Model ID for abstractive summarization",
+    )
+    max_summary_tokens: int = Field(
+        default=200,
+        ge=50,
+        le=1000,
+        description="Maximum tokens for LLM summary responses",
+    )
+    max_facts: int = Field(
+        default=20,
+        ge=1,
+        le=100,
+        description="Maximum extracted key facts for extractive mode",
+    )
+    anchor_length: int = Field(
+        default=150,
+        ge=50,
+        le=500,
+        description="Character length for each extractive anchor",
+    )
+
+    @model_validator(mode="after")
+    def _validate_model_when_enabled(self) -> Self:
+        """Require a summarization model when dual-mode is enabled."""
+        if self.enabled and not self.summarization_model.strip():
+            msg = "summarization_model must be non-blank when dual-mode is enabled"
+            raise ValueError(msg)
+        return self
+
+
 class ArchivalConfig(BaseModel):
     """Archival configuration.
 
     Attributes:
         enabled: Whether archival is enabled.
         age_threshold_days: Minimum age in days before archival.
+        dual_mode: Dual-mode archival configuration.
     """
 
     model_config = ConfigDict(frozen=True, allow_inf_nan=False)
@@ -67,6 +132,10 @@ class ArchivalConfig(BaseModel):
         ge=1,
         description="Minimum age in days before archival",
     )
+    dual_mode: DualModeConfig = Field(
+        default_factory=DualModeConfig,
+        description="Dual-mode archival configuration",
+    )
 
 
 class ConsolidationConfig(BaseModel):