Plastic Labs Internal Wiki

Recently Updated

dreaming-enhancements

29 min read

Dreaming Enhancements

Status: Draft | Owner: vineeth | Last updated: 2026-03-25

1. Problem Statement

Honcho’s dreamer currently runs a single dream type (omni) that executes a fixed two-phase cycle: deduction specialist followed by induction specialist. While this produces useful observations, it has several limitations:

  1. No semantic organization of conclusions. Observations (documents) accumulate without any categorical structure. There is no mechanism for grouping related conclusions, tracking thematic areas, or enabling developers to query conclusions by topic. The only filtering axes today are level (explicit/deductive/inductive/contradiction), observer, observed, and free-text semantic search.

  2. No developer-defined investigations. The dreamer only runs its built-in deduction+induction cycle. Developers cannot define custom recurring analyses (e.g., “summarize this user’s communication preferences weekly” or “maintain a running personality profile”). There is no way to produce scoped, structured outputs beyond individual observations.

  3. Weak consolidation. The deduction specialist can delete outdated observations and the induction specialist creates patterns, but there is no systematic approach to deduplication, staleness detection, or capacity management. As observation counts grow, the system accumulates redundant and stale entries without pressure to clean them up.

These gaps prevent Honcho from being a truly agentic, self-evolving memory environment. The dreamer should not just create observations — it should organize, curate, and produce structured outputs that grow more useful over time.

2. Goals / Non-Goals

Goals

  • G1: Agent-generated tagging. The dreaming agent autonomously discovers and applies semantic tags to conclusions based on patterns it observes. Tags are stored in the new public metadata JSONB column on the documents table (from the conclusion-tagging spec). Limits on per-conclusion tag count and total tag vocabulary prevent unbounded growth.

  • G2: Materialized dreams. Developers can define recurring or one-time investigations (dream definitions) that produce scoped documents — prose summaries, conclusion lists, structured analyses. A new dream_definitions table tracks definitions, cadence, and latest results. Results are stored as conclusions with metadata linking them to their definition.

  • G3: Improved consolidation. Smarter deduplication via embedding similarity, staleness detection based on age and derivation activity, and capacity-based pressure that triggers cleanup when observation counts exceed thresholds.

  • G4: New DreamType variants. Extend the DreamType enum beyond omni to support tag, materialize, and consolidate as independently schedulable dream phases.

Non-Goals

  • NG1: File system primitives. Materialized dream results are stored as conclusions or in metadata, not as files. File-based output is a future concern.
  • NG2: User-facing tag management UI. Tags are created and managed by the dreaming agent, not by end users directly through the API (though they are readable via the public metadata field).
  • NG3: Real-time tagging during message ingestion. Tags are applied during dream cycles, not during the deriver’s representation pipeline.
  • NG4: Cross-workspace tag taxonomies. Tags are scoped to a single workspace and observer/observed pair.

3. Design

3.1 Agent-Generated Tagging

3.1.1 Overview

A new TaggingSpecialist runs as an optional third phase of the dream cycle. It examines existing observations and applies semantic tags based on patterns it discovers. Tags are arbitrary strings chosen by the agent — Honcho does not impose a fixed taxonomy. The agent organically evolves the tag vocabulary as it learns more about a peer.

3.1.2 Tag Schema and Limits

Tags are stored in the metadata JSONB column on the documents table (the public, read-write column being added by the conclusion-tagging spec, distinct from internal_metadata).

Storage format:

{
  "tags": ["communication-style", "career", "family"],
  "tagged_at": "2026-03-25T12:00:00Z",
  "tagged_by_run": "a1b2c3d4"
}

Limits (enforced in the tool executor, not by DB constraints):

LimitValueRationale
Max tags per conclusion5Prevents over-tagging; forces prioritization
Max distinct tags per collection50Bounds vocabulary growth per observer/observed pair
Max tag length64 charsPrevents verbose tag names
Tag formatlowercase alphanumeric + hyphensNormalizes for consistent matching

When the agent attempts to exceed the collection-level tag limit (50), the tool executor returns an error instructing the agent to reuse existing tags or consolidate similar ones.

Config additions to DreamSettings:

# In DreamSettings
TAGGING_MODEL: str = "claude-haiku-4-5"
MAX_TAGS_PER_CONCLUSION: Annotated[int, Field(default=5, gt=0, le=20)] = 5
MAX_TAGS_PER_COLLECTION: Annotated[int, Field(default=50, gt=0, le=200)] = 50
MAX_TAG_LENGTH: Annotated[int, Field(default=64, gt=0, le=128)] = 64

3.1.3 TaggingSpecialist

The TaggingSpecialist extends BaseSpecialist and is registered in the SPECIALISTS dict.

File: src/dreamer/specialists.py

class TaggingSpecialist(BaseSpecialist):
    """
    Applies semantic tags to observations based on discovered patterns.
 
    This specialist:
    1. Reviews recent and high-value observations
    2. Identifies thematic categories across the observation space
    3. Applies tags to observations via the tag_observations tool
    4. Manages the tag vocabulary (merges, renames)
    """
 
    name: str = "tagging"
    peer_card_update_instruction: str = ""  # Tagging specialist does not update peer card
 
    def get_tools(self, *, peer_card_enabled: bool = True) -> list[dict[str, Any]]:
        return TAGGING_SPECIALIST_TOOLS
 
    def get_model(self) -> str:
        return settings.DREAM.TAGGING_MODEL
 
    def get_max_tokens(self) -> int:
        return 8192
 
    def get_max_iterations(self) -> int:
        return 10
 
    def build_system_prompt(self, observed: str, *, peer_card_enabled: bool = True) -> str:
        # See section 3.1.4 for full prompt
        ...
 
    def build_user_prompt(self, hints: list[str] | None, peer_card: list[str] | None = None) -> str:
        ...

Registration:

SPECIALISTS: dict[str, BaseSpecialist] = {
    "deduction": DeductionSpecialist(),
    "induction": InductionSpecialist(),
    "tagging": TaggingSpecialist(),
}

3.1.4 TaggingSpecialist System Prompt (Sketch)

You are a semantic tagging agent organizing observations about {observed}.

## YOUR JOB

Apply semantic tags to observations to create a navigable knowledge structure.
Tags should reflect the CONTENT THEMES you discover, not the observation level.

## GUIDELINES

- Tags are lowercase, hyphenated strings (e.g., "career-goals", "family-dynamics", "communication-style")
- Each observation can have at most {max_tags_per_conclusion} tags
- The total vocabulary is capped at {max_tags_per_collection} distinct tags per collection
- Reuse existing tags when possible -- consistency matters more than precision
- Merge near-duplicate tags (e.g., "work" and "career" should be one tag)

## PHASE 1: DISCOVERY

Explore the observation space to understand what themes exist:
- `get_recent_observations` - See recent observations
- `search_memory` - Search by topic
- `get_existing_tags` - See what tags already exist and their counts

## PHASE 2: TAGGING

Apply tags to untagged or under-tagged observations:
- `tag_observations` - Apply tags to one or more observations
- `rename_tag` - Rename a tag across all observations that have it
- `merge_tags` - Merge two tags into one

## RULES

1. Focus on untagged observations first
2. Use broad, reusable categories -- not observation-specific labels
3. Every tag should apply to at least 2 observations
4. Do not create tags that duplicate the observation level (no "deductive" tag)
5. Prefer fewer, well-applied tags over many sparse ones

3.1.5 Tagging Tools

New tools added to src/utils/agent_tools.py:

tag_observations — Apply tags to observations by ID.

TOOLS["tag_observations"] = {
    "name": "tag_observations",
    "description": "Apply semantic tags to one or more observations. Tags are lowercase hyphenated strings.",
    "input_schema": {
        "type": "object",
        "properties": {
            "operations": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "observation_id": {
                            "type": "string",
                            "description": "Document ID of the observation to tag",
                        },
                        "tags": {
                            "type": "array",
                            "items": {"type": "string"},
                            "description": "Tags to apply (lowercase, hyphenated, max 64 chars each)",
                        },
                    },
                    "required": ["observation_id", "tags"],
                },
            },
        },
        "required": ["operations"],
    },
}

get_existing_tags — Retrieve the current tag vocabulary with counts.

TOOLS["get_existing_tags"] = {
    "name": "get_existing_tags",
    "description": "Get all existing tags and their usage counts for this collection.",
    "input_schema": {
        "type": "object",
        "properties": {},
    },
}

rename_tag — Rename a tag across all observations.

TOOLS["rename_tag"] = {
    "name": "rename_tag",
    "description": "Rename a tag across all observations that have it.",
    "input_schema": {
        "type": "object",
        "properties": {
            "old_name": {"type": "string", "description": "Current tag name"},
            "new_name": {"type": "string", "description": "New tag name"},
        },
        "required": ["old_name", "new_name"],
    },
}

merge_tags — Merge two tags into one.

TOOLS["merge_tags"] = {
    "name": "merge_tags",
    "description": "Merge two tags into one. All observations with either tag will have the target tag.",
    "input_schema": {
        "type": "object",
        "properties": {
            "source_tag": {"type": "string", "description": "Tag to merge from (will be removed)"},
            "target_tag": {"type": "string", "description": "Tag to merge into (will be kept)"},
        },
        "required": ["source_tag", "target_tag"],
    },
}

Tool list constant:

TAGGING_SPECIALIST_TOOLS: list[dict[str, Any]] = [
    TOOLS["get_recent_observations"],
    TOOLS["search_memory"],
    TOOLS["get_existing_tags"],
    TOOLS["tag_observations"],
    TOOLS["rename_tag"],
    TOOLS["merge_tags"],
]

3.1.6 Tool Executor Implementation

The tag_observations handler in create_tool_executor:

  1. Validates each tag against the format regex ^[a-z0-9]+(-[a-z0-9]+)*$ and length limit.
  2. Checks that the observation exists and belongs to the correct workspace/observer/observed.
  3. Checks that applying the tags would not exceed MAX_TAGS_PER_CONCLUSION for any individual observation.
  4. Queries the current distinct tag count for the collection. If adding new tags would exceed MAX_TAGS_PER_COLLECTION, returns an error with the current vocabulary so the agent can reuse existing tags.
  5. Updates the metadata JSONB column: metadata = jsonb_set(metadata, '{tags}', new_tags_array). Also sets metadata.tagged_at and metadata.tagged_by_run.
  6. Returns a summary of applied tags.

The get_existing_tags handler:

  1. Queries: SELECT DISTINCT jsonb_array_elements_text(metadata->'tags') AS tag, COUNT(*) FROM documents WHERE workspace_name = :ws AND observer = :obs AND observed = :obd AND deleted_at IS NULL AND metadata->'tags' IS NOT NULL GROUP BY tag ORDER BY count DESC.
  2. Returns the tag vocabulary with counts.

The rename_tag and merge_tags handlers perform bulk updates using JSONB array manipulation:

-- rename_tag: replace old_name with new_name in all documents that have old_name
UPDATE documents
SET metadata = jsonb_set(
    metadata,
    '{tags}',
    (
        SELECT jsonb_agg(DISTINCT CASE WHEN elem = :old_name THEN :new_name ELSE elem END)
        FROM jsonb_array_elements_text(metadata->'tags') AS elem
    )
)
WHERE workspace_name = :ws
  AND observer = :obs AND observed = :obd
  AND deleted_at IS NULL
  AND metadata->'tags' @> to_jsonb(:old_name::text);

3.1.7 How Tags Influence Future Reasoning

Tags flow into the existing specialist workflows via two mechanisms:

  1. Enriched observation display. When get_recent_observations or search_memory returns results, observations that have tags include them in the formatted output: [id:abc123] [tags: career, goals] Content of the observation.... This gives deduction and induction specialists thematic context without additional tool calls.

  2. Tag-filtered search. A new optional tag parameter is added to search_memory:

"tag": {
    "type": "string",
    "description": "(Optional) Filter results to observations with this tag",
}

The search_memory handler filters results by checking metadata->'tags' @> to_jsonb(:tag::text) before or after the vector similarity search.

  1. Materialized dream scoping. Dream definitions (section 3.2) can specify tags to scope their investigation, allowing focused analyses of specific thematic areas.

3.2 Materialized Dreams

3.2.1 Overview

Materialized dreams are developer-defined investigations that produce scoped output documents. Unlike the built-in omni dream cycle, materialized dreams are custom tasks with:

  • A definition (what to investigate, what output to produce)
  • A cadence (how often to run)
  • A scope (which observer/observed pair, optional tag filter)
  • A result (stored as conclusions with metadata linking to the definition)

3.2.2 dream_definitions Table

DDL:

CREATE TABLE dream_definitions (
    id          TEXT PRIMARY KEY DEFAULT nanoid(),
    name        TEXT NOT NULL,
    description TEXT,
 
    -- Scope
    workspace_name  TEXT NOT NULL REFERENCES workspaces(name),
    observer        TEXT,       -- NULL = all observers
    observed        TEXT,       -- NULL = all observed peers
 
    -- Investigation parameters
    instructions    TEXT NOT NULL,           -- System prompt for the investigation agent
    output_type     TEXT NOT NULL DEFAULT 'conclusions',  -- 'conclusions' | 'metadata' | 'prose'
    tag_filter      TEXT[],                 -- Optional: only examine observations with these tags
    max_output_items INT DEFAULT 10,        -- Max conclusions/items to produce per run
 
    -- Scheduling
    trigger_type    TEXT NOT NULL DEFAULT 'on_demand',  -- 'cron' | 'on_change' | 'on_demand'
    cron_expression TEXT,       -- Required if trigger_type = 'cron' (e.g., '0 0 * * 0' = weekly)
    min_interval_hours  INT DEFAULT 24,     -- Minimum hours between runs (prevents spam)
 
    -- State
    enabled         BOOLEAN NOT NULL DEFAULT true,
    last_run_at     TIMESTAMPTZ,
    last_run_id     TEXT,       -- Correlates with DreamRunEvent.run_id
    last_run_status TEXT,       -- 'success' | 'failure' | 'running'
    run_count       INT NOT NULL DEFAULT 0,
 
    -- Metadata
    h_metadata      JSONB NOT NULL DEFAULT '{}'::jsonb,
    internal_metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
 
    -- Constraints
    CONSTRAINT dream_definitions_id_length CHECK (length(id) = 21),
    CONSTRAINT dream_definitions_id_format CHECK (id ~ '^[A-Za-z0-9_-]+$'),
    CONSTRAINT dream_definitions_name_length CHECK (length(name) <= 256),
    CONSTRAINT dream_definitions_instructions_length CHECK (length(instructions) <= 65535),
    CONSTRAINT dream_definitions_trigger_type_valid CHECK (trigger_type IN ('cron', 'on_change', 'on_demand')),
    CONSTRAINT dream_definitions_output_type_valid CHECK (output_type IN ('conclusions', 'metadata', 'prose')),
    CONSTRAINT dream_definitions_cron_required CHECK (
        trigger_type != 'cron' OR cron_expression IS NOT NULL
    ),
 
    -- Foreign keys for observer/observed (when specified)
    FOREIGN KEY (observer, workspace_name) REFERENCES peers(name, workspace_name),
    FOREIGN KEY (observed, workspace_name) REFERENCES peers(name, workspace_name)
);
 
-- Index for finding definitions that need to run
CREATE INDEX ix_dream_definitions_trigger_lookup
    ON dream_definitions (workspace_name, trigger_type, enabled, last_run_at);
 
-- Unique name per workspace
CREATE UNIQUE INDEX uq_dream_definitions_name_workspace
    ON dream_definitions (workspace_name, name);

SQLAlchemy model (added to src/models.py):

@final
class DreamDefinition(Base):
    __tablename__: str = "dream_definitions"
 
    id: Mapped[str] = mapped_column(TEXT, default=generate_nanoid, primary_key=True)
    name: Mapped[str] = mapped_column(TEXT, nullable=False)
    description: Mapped[str | None] = mapped_column(TEXT, nullable=True)
 
    workspace_name: Mapped[str] = mapped_column(
        ForeignKey("workspaces.name"), nullable=False, index=True
    )
    observer: Mapped[str | None] = mapped_column(TEXT, nullable=True)
    observed: Mapped[str | None] = mapped_column(TEXT, nullable=True)
 
    instructions: Mapped[str] = mapped_column(TEXT, nullable=False)
    output_type: Mapped[str] = mapped_column(
        TEXT, nullable=False, server_default="conclusions"
    )
    tag_filter: Mapped[list[str] | None] = mapped_column(
        ARRAY(TEXT), nullable=True
    )
    max_output_items: Mapped[int] = mapped_column(
        Integer, nullable=False, server_default=text("10")
    )
 
    trigger_type: Mapped[str] = mapped_column(
        TEXT, nullable=False, server_default="on_demand"
    )
    cron_expression: Mapped[str | None] = mapped_column(TEXT, nullable=True)
    min_interval_hours: Mapped[int] = mapped_column(
        Integer, nullable=False, server_default=text("24")
    )
 
    enabled: Mapped[bool] = mapped_column(
        Boolean, nullable=False, server_default=text("true")
    )
    last_run_at: Mapped[datetime.datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )
    last_run_id: Mapped[str | None] = mapped_column(TEXT, nullable=True)
    last_run_status: Mapped[str | None] = mapped_column(TEXT, nullable=True)
    run_count: Mapped[int] = mapped_column(
        Integer, nullable=False, server_default=text("0")
    )
 
    h_metadata: Mapped[dict[str, Any]] = mapped_column(
        "metadata", JSONB, default=dict, server_default=text("'{}'::jsonb")
    )
    internal_metadata: Mapped[dict[str, Any]] = mapped_column(
        "internal_metadata", JSONB, default=dict, server_default=text("'{}'::jsonb")
    )
    created_at: Mapped[datetime.datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), index=True
    )
    updated_at: Mapped[datetime.datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
    )
 
    __table_args__ = (
        UniqueConstraint("workspace_name", "name"),
        CheckConstraint("length(id) = 21", name="dream_def_id_length"),
        CheckConstraint("id ~ '^[A-Za-z0-9_-]+$'", name="dream_def_id_format"),
        CheckConstraint("length(name) <= 256", name="dream_def_name_length"),
        CheckConstraint("length(instructions) <= 65535", name="dream_def_instructions_length"),
        CheckConstraint(
            "trigger_type IN ('cron', 'on_change', 'on_demand')",
            name="dream_def_trigger_type_valid",
        ),
        CheckConstraint(
            "output_type IN ('conclusions', 'metadata', 'prose')",
            name="dream_def_output_type_valid",
        ),
        CheckConstraint(
            "trigger_type != 'cron' OR cron_expression IS NOT NULL",
            name="dream_def_cron_required",
        ),
        ForeignKeyConstraint(
            ["observer", "workspace_name"],
            ["peers.name", "peers.workspace_name"],
        ),
        ForeignKeyConstraint(
            ["observed", "workspace_name"],
            ["peers.name", "peers.workspace_name"],
        ),
        Index(
            "ix_dream_definitions_trigger_lookup",
            "workspace_name", "trigger_type", "enabled", "last_run_at",
        ),
    )

3.2.3 Trigger Types

TypeDescriptionScheduling mechanism
cronRuns on a cron scheduleDreamScheduler polls dream_definitions on a configurable interval (default 60s). When a cron-triggered definition is due (based on cron_expression and last_run_at), it enqueues a materialize dream task.
on_changeRuns when new observations are created for the scoped observer/observed pairAfter the deriver creates observations, it checks for on_change definitions matching the scope and enqueues a materialize dream if min_interval_hours has elapsed since last_run_at.
on_demandRuns only when triggered via APINo automatic scheduling. Triggered by POST /workspaces/{id}/dream_definitions/{def_id}/trigger.

3.2.4 Output Types

TypeStorageDescription
conclusionsNew Document rows with level='inductive'The investigation agent creates observations via create_observations. Each observation’s metadata includes {"dream_definition_id": "<def_id>", "dream_run_id": "<run_id>"}.
metadataWritten to dream_definitions.internal_metadata.last_resultThe investigation agent returns structured JSON that is stored on the definition row itself. Useful for compact results that don’t need vector search.
proseNew Document row with level='inductive' and metadata.output_type='prose'A single long-form document summarizing the investigation. Stored as one observation with the full prose content.

3.2.5 API Surface

New CRUD endpoints under /workspaces/{workspace_id}/dream_definitions:

Schemas (added to src/schemas/api.py):

class DreamDefinitionCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=256)
    description: str | None = None
    observer: str | None = Field(None, description="Observer peer name (null = all)")
    observed: str | None = Field(None, description="Observed peer name (null = all)")
    instructions: str = Field(..., min_length=1, max_length=65535)
    output_type: Literal["conclusions", "metadata", "prose"] = "conclusions"
    tag_filter: list[str] | None = None
    max_output_items: int = Field(default=10, ge=1, le=100)
    trigger_type: Literal["cron", "on_change", "on_demand"] = "on_demand"
    cron_expression: str | None = None
    min_interval_hours: int = Field(default=24, ge=1, le=720)
 
class DreamDefinitionUpdate(BaseModel):
    name: str | None = None
    description: str | None = None
    instructions: str | None = None
    output_type: Literal["conclusions", "metadata", "prose"] | None = None
    tag_filter: list[str] | None = None
    max_output_items: int | None = None
    trigger_type: Literal["cron", "on_change", "on_demand"] | None = None
    cron_expression: str | None = None
    min_interval_hours: int | None = None
    enabled: bool | None = None
 
class DreamDefinitionResponse(BaseModel):
    id: str
    name: str
    description: str | None
    workspace_name: str
    observer: str | None
    observed: str | None
    instructions: str
    output_type: str
    tag_filter: list[str] | None
    max_output_items: int
    trigger_type: str
    cron_expression: str | None
    min_interval_hours: int
    enabled: bool
    last_run_at: datetime.datetime | None
    last_run_id: str | None
    last_run_status: str | None
    run_count: int
    created_at: datetime.datetime
    updated_at: datetime.datetime
    metadata: dict[str, Any] = Field(default_factory=dict)
 
    model_config = ConfigDict(from_attributes=True, populate_by_name=True)

Endpoints (new router file src/routers/dream_definitions.py):

MethodPathDescription
POST/workspaces/{workspace_id}/dream_definitionsCreate a dream definition
GET/workspaces/{workspace_id}/dream_definitionsList dream definitions (paginated)
GET/workspaces/{workspace_id}/dream_definitions/{definition_id}Get a single definition
PUT/workspaces/{workspace_id}/dream_definitions/{definition_id}Update a definition
DELETE/workspaces/{workspace_id}/dream_definitions/{definition_id}Delete a definition
POST/workspaces/{workspace_id}/dream_definitions/{definition_id}/triggerManually trigger a materialized dream run

The trigger endpoint:

@router.post(
    "/{definition_id}/trigger",
    status_code=202,
    response_model=DreamTriggerResponse,
)
async def trigger_dream_definition(
    workspace_id: str = Path(...),
    definition_id: str = Path(...),
    db: AsyncSession = db,
) -> DreamTriggerResponse:
    """
    Manually trigger a materialized dream for this definition.
 
    Returns 202 Accepted with a run_id for tracking.
    Respects min_interval_hours -- returns 429 if too soon.
    """
    definition = await crud.get_dream_definition(db, workspace_id, definition_id)
    if definition is None:
        raise ResourceNotFoundException("Dream definition not found")
 
    if not definition.enabled:
        raise ValidationException("Dream definition is disabled")
 
    # Check min_interval_hours
    if definition.last_run_at:
        hours_since = (datetime.now(timezone.utc) - definition.last_run_at).total_seconds() / 3600
        if hours_since < definition.min_interval_hours:
            raise HTTPException(
                status_code=429,
                detail=f"Too soon. {definition.min_interval_hours - hours_since:.1f} hours remaining."
            )
 
    run_id = await enqueue_materialized_dream(workspace_id, definition)
    return DreamTriggerResponse(run_id=run_id, status="accepted")

3.2.6 DreamRunner Execution Flow

When a materialize dream task is dequeued, process_dream in orchestrator.py routes to a new run_materialized_dream function:

process_dream(payload)
  |
  +--> match payload.dream_type:
         case DreamType.OMNI:      run_dream(...)          # existing
         case DreamType.TAG:       run_tagging(...)         # new (section 3.1)
         case DreamType.MATERIALIZE: run_materialized_dream(...)  # new
         case DreamType.CONSOLIDATE: run_consolidation(...)  # new (section 3.3)

run_materialized_dream flow:

  1. Load the DreamDefinition from the database using the definition_id in the payload.
  2. Mark last_run_status = 'running' and last_run_at = now().
  3. Build a MaterializationSpecialist dynamically:
    • System prompt is constructed from definition.instructions plus a standard preamble that explains the output format requirements.
    • Tools are a subset: get_recent_observations, search_memory, search_messages, create_observations (if output_type is conclusions or prose).
    • If tag_filter is set, the search tools are pre-filtered to only return observations with matching tags.
  4. Run the specialist via BaseSpecialist.run(...).
  5. If output_type == 'metadata', parse the agent’s final response as JSON and write to definition.internal_metadata['last_result'].
  6. Update last_run_status = 'success', increment run_count.
  7. On failure, set last_run_status = 'failure' and store error in internal_metadata['last_error'].

Payload extension:

class MaterializeDreamPayload(DreamPayload):
    """Extended payload for materialized dreams."""
    dream_type: Literal["materialize"] = "materialize"
    definition_id: str  # References dream_definitions.id

The DreamPayload model in src/utils/queue_payload.py gains an optional definition_id field:

class DreamPayload(BasePayload):
    task_type: Literal["dream"] = "dream"
    dream_type: DreamType
    observer: str
    observed: str
    session_name: str | None = None
    definition_id: str | None = None  # NEW: for materialized dreams

3.2.7 Cron Polling

The DreamScheduler gains a new periodic task that polls for due cron definitions:

async def _poll_cron_definitions(self) -> None:
    """Check for cron-triggered dream definitions that are due to run."""
    while True:
        try:
            async with tracked_db("dream_cron_poll") as db:
                due_definitions = await crud.get_due_cron_definitions(db)
                for defn in due_definitions:
                    await enqueue_materialized_dream(
                        defn.workspace_name, defn
                    )
        except Exception as e:
            logger.error("Cron poll failed: %s", e)
        await asyncio.sleep(self.cron_poll_interval_seconds)

Config addition:

# In DreamSettings
CRON_POLL_INTERVAL_SECONDS: Annotated[int, Field(default=60, gt=10, le=3600)] = 60

The crud.get_due_cron_definitions query:

SELECT * FROM dream_definitions
WHERE enabled = true
  AND trigger_type = 'cron'
  AND (last_run_at IS NULL
       OR last_run_at + (min_interval_hours || ' hours')::interval < now())
  AND last_run_status != 'running'
ORDER BY last_run_at ASC NULLS FIRST
LIMIT 10;

Cron expression evaluation uses the croniter library (add to pyproject.toml dependencies) to check if the definition is due based on cron_expression and last_run_at.

3.3 Consolidation Improvements

3.3.1 Overview

A new ConsolidationSpecialist (or enhanced consolidation mode) systematically addresses three problems: duplicate observations, stale observations, and unbounded growth.

3.3.2 Smarter Deduplication

Current behavior: The deduction specialist can manually identify and delete duplicates during its normal run, but has no systematic mechanism for finding them.

New behavior: A dedicated deduplication phase uses embedding similarity to find near-duplicate observations.

Implementation — new tool find_near_duplicates:

TOOLS["find_near_duplicates"] = {
    "name": "find_near_duplicates",
    "description": "Find observations that are semantically very similar to each other (potential duplicates). Returns pairs of observations with their similarity scores.",
    "input_schema": {
        "type": "object",
        "properties": {
            "similarity_threshold": {
                "type": "number",
                "description": "Minimum cosine similarity to consider as near-duplicate (default: 0.92, range: 0.80-0.99)",
                "default": 0.92,
            },
            "limit": {
                "type": "integer",
                "description": "Maximum number of duplicate pairs to return (default: 20)",
                "default": 20,
            },
        },
    },
}

Tool executor implementation:

The handler executes a self-join query on the documents table using pgvector’s cosine distance:

SELECT
    d1.id AS id_a, d1.content AS content_a, d1.level AS level_a, d1.created_at AS created_a,
    d2.id AS id_b, d2.content AS content_b, d2.level AS level_b, d2.created_at AS created_b,
    1 - (d1.embedding <=> d2.embedding) AS similarity
FROM documents d1
JOIN documents d2
    ON d1.workspace_name = d2.workspace_name
    AND d1.observer = d2.observer
    AND d1.observed = d2.observed
    AND d1.id < d2.id  -- avoid self-pairs and duplicates
WHERE d1.workspace_name = :ws
  AND d1.observer = :obs
  AND d1.observed = :obd
  AND d1.deleted_at IS NULL
  AND d2.deleted_at IS NULL
  AND 1 - (d1.embedding <=> d2.embedding) >= :threshold
ORDER BY similarity DESC
LIMIT :limit;

The consolidation specialist receives these pairs and decides which to merge (keeping the higher-level or more-derived observation) and which to delete.

3.3.3 Staleness Detection

New tool find_stale_observations:

TOOLS["find_stale_observations"] = {
    "name": "find_stale_observations",
    "description": "Find observations that may be stale based on age and lack of recent derivation activity. Returns observations ordered by staleness score (most stale first).",
    "input_schema": {
        "type": "object",
        "properties": {
            "max_age_days": {
                "type": "integer",
                "description": "Observations older than this are candidates for staleness review (default: 90)",
                "default": 90,
            },
            "limit": {
                "type": "integer",
                "description": "Maximum number of stale observations to return (default: 20)",
                "default": 20,
            },
        },
    },
}

Staleness scoring:

An observation’s staleness score is computed as:

staleness = age_days / max_age_days * (1 / max(times_derived, 1))

Observations that are old AND have low times_derived counts are the most stale. The tool returns these for the consolidation specialist to review and decide whether to soft-delete.

The query:

SELECT id, content, level, created_at, times_derived,
    EXTRACT(EPOCH FROM (now() - created_at)) / 86400.0 AS age_days,
    (EXTRACT(EPOCH FROM (now() - created_at)) / 86400.0 / :max_age_days)
        * (1.0 / GREATEST(times_derived, 1)) AS staleness_score
FROM documents
WHERE workspace_name = :ws
  AND observer = :obs AND observed = :obd
  AND deleted_at IS NULL
  AND created_at < now() - (:max_age_days || ' days')::interval
ORDER BY staleness_score DESC
LIMIT :limit;

The consolidation specialist reviews stale observations and can:

  • Delete those that are clearly outdated or superseded
  • Bump times_derived for those that are still relevant (re-confirming them)
  • Create updated replacement observations when the information has changed

3.3.4 Capacity-Based Pressure

When the total observation count for a collection exceeds a configurable threshold, the consolidation specialist receives pressure signals to clean up.

Config additions:

# In DreamSettings
CONSOLIDATION_MODEL: str = "claude-haiku-4-5"
CONSOLIDATION_SOFT_LIMIT: Annotated[int, Field(default=500, gt=50, le=10000)] = 500
CONSOLIDATION_HARD_LIMIT: Annotated[int, Field(default=1000, gt=100, le=50000)] = 1000

Behavior:

  • Below CONSOLIDATION_SOFT_LIMIT: Consolidation is optional. The specialist runs normally with find_near_duplicates and find_stale_observations as discovery tools.
  • Between soft and hard limits: Consolidation is pressured. The system prompt includes a directive: “The observation space is growing large ({count}/{hard_limit}). Prioritize deduplication and staleness cleanup.”
  • At or above CONSOLIDATION_HARD_LIMIT: Consolidation is urgent. The specialist receives a strong directive to reduce the observation count. The find_near_duplicates threshold is lowered automatically (e.g., 0.85 instead of 0.92) to catch more potential merges.

The observation count is fetched at the start of the consolidation run:

count_stmt = select(func.count(Document.id)).where(
    Document.workspace_name == workspace_name,
    Document.observer == observer,
    Document.observed == observed,
    Document.deleted_at.is_(None),
)
observation_count = await db.scalar(count_stmt)

3.3.5 ConsolidationSpecialist

class ConsolidationSpecialist(BaseSpecialist):
    """
    Consolidates the observation space by deduplicating, removing stale
    observations, and managing capacity.
    """
 
    name: str = "consolidation"
    peer_card_update_instruction: str = ""
 
    def get_tools(self, *, peer_card_enabled: bool = True) -> list[dict[str, Any]]:
        return CONSOLIDATION_SPECIALIST_TOOLS
 
    def get_model(self) -> str:
        return settings.DREAM.CONSOLIDATION_MODEL
 
    def get_max_tokens(self) -> int:
        return 8192
 
    def get_max_iterations(self) -> int:
        return 15
 
    def build_system_prompt(self, observed: str, *, peer_card_enabled: bool = True) -> str:
        ...  # See below
 
    def build_user_prompt(
        self,
        hints: list[str] | None,
        peer_card: list[str] | None = None,
    ) -> str:
        ...

System prompt sketch:

You are a memory consolidation agent for observations about {observed}.

## YOUR JOB

Clean up and optimize the observation space by:
1. Finding and merging near-duplicate observations
2. Identifying and removing stale, outdated observations
3. Ensuring the observation space stays within capacity limits

## TOOLS

- `find_near_duplicates` - Find pairs of very similar observations
- `find_stale_observations` - Find old, unreinforced observations
- `get_recent_observations` - See recent observations for context
- `search_memory` - Search for specific topics
- `delete_observations` - Remove observations (soft delete)
- `create_observations` - Create merged/updated observations when consolidating

## DEDUPLICATION STRATEGY

When you find near-duplicates:
1. Keep the observation with MORE source_ids (better provenance)
2. If equal, keep the NEWER one (more likely to reflect current state)
3. If one is higher level (inductive > deductive > explicit), prefer it
4. Delete the other one
5. If both contain unique information, create a merged observation and delete both originals

## STALENESS STRATEGY

When reviewing stale observations:
1. Check if the observation is still likely true (use search_memory for context)
2. If clearly outdated (e.g., refers to a past event as upcoming), delete it
3. If still relevant but old, leave it -- age alone is not grounds for deletion
4. If superseded by a newer observation with the same information, delete the old one

{capacity_pressure_directive}

## RULES

1. Never delete an observation without checking if other observations depend on it (via source_ids)
2. When in doubt, keep the observation -- false deletion is worse than clutter
3. Log what you delete and why in your final summary
4. Prefer merging over deletion when both observations add value

Tool list:

CONSOLIDATION_SPECIALIST_TOOLS: list[dict[str, Any]] = [
    TOOLS["find_near_duplicates"],
    TOOLS["find_stale_observations"],
    TOOLS["get_recent_observations"],
    TOOLS["search_memory"],
    TOOLS["delete_observations"],
    TOOLS["create_observations"],
    TOOLS["get_reasoning_chain"],
]

4. Migration Plan

4.1 Database Migration (Alembic)

Migration 1: Add metadata column to documents table.

This migration is owned by the conclusion-tagging spec and is a prerequisite. It adds the public metadata JSONB column (mapped as h_metadata in SQLAlchemy, serialized as metadata in the DB) to the documents table alongside the existing internal_metadata.

Migration 2: Create dream_definitions table.

def upgrade() -> None:
    op.create_table(
        "dream_definitions",
        sa.Column("id", sa.TEXT(), primary_key=True),
        sa.Column("name", sa.TEXT(), nullable=False),
        sa.Column("description", sa.TEXT(), nullable=True),
        sa.Column("workspace_name", sa.TEXT(), sa.ForeignKey("workspaces.name"), nullable=False),
        sa.Column("observer", sa.TEXT(), nullable=True),
        sa.Column("observed", sa.TEXT(), nullable=True),
        sa.Column("instructions", sa.TEXT(), nullable=False),
        sa.Column("output_type", sa.TEXT(), nullable=False, server_default="conclusions"),
        sa.Column("tag_filter", sa.ARRAY(sa.TEXT()), nullable=True),
        sa.Column("max_output_items", sa.Integer(), nullable=False, server_default="10"),
        sa.Column("trigger_type", sa.TEXT(), nullable=False, server_default="on_demand"),
        sa.Column("cron_expression", sa.TEXT(), nullable=True),
        sa.Column("min_interval_hours", sa.Integer(), nullable=False, server_default="24"),
        sa.Column("enabled", sa.Boolean(), nullable=False, server_default="true"),
        sa.Column("last_run_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("last_run_id", sa.TEXT(), nullable=True),
        sa.Column("last_run_status", sa.TEXT(), nullable=True),
        sa.Column("run_count", sa.Integer(), nullable=False, server_default="0"),
        sa.Column("metadata", postgresql.JSONB(), nullable=False, server_default="{}"),
        sa.Column("internal_metadata", postgresql.JSONB(), nullable=False, server_default="{}"),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
    )
 
    # Constraints
    op.create_check_constraint("dream_def_id_length", "dream_definitions", "length(id) = 21")
    op.create_check_constraint("dream_def_id_format", "dream_definitions", "id ~ '^[A-Za-z0-9_-]+$'")
    op.create_check_constraint("dream_def_name_length", "dream_definitions", "length(name) <= 256")
    op.create_check_constraint("dream_def_instructions_length", "dream_definitions", "length(instructions) <= 65535")
    op.create_check_constraint("dream_def_trigger_type_valid", "dream_definitions", "trigger_type IN ('cron', 'on_change', 'on_demand')")
    op.create_check_constraint("dream_def_output_type_valid", "dream_definitions", "output_type IN ('conclusions', 'metadata', 'prose')")
    op.create_check_constraint("dream_def_cron_required", "dream_definitions", "trigger_type != 'cron' OR cron_expression IS NOT NULL")
 
    # Foreign key constraints for observer/observed
    op.create_foreign_key(
        "fk_dream_def_observer", "dream_definitions",
        "peers", ["observer", "workspace_name"], ["name", "workspace_name"],
    )
    op.create_foreign_key(
        "fk_dream_def_observed", "dream_definitions",
        "peers", ["observed", "workspace_name"], ["name", "workspace_name"],
    )
 
    # Indexes
    op.create_index(
        "ix_dream_definitions_trigger_lookup", "dream_definitions",
        ["workspace_name", "trigger_type", "enabled", "last_run_at"],
    )
    op.create_unique_constraint(
        "uq_dream_definitions_name_workspace", "dream_definitions",
        ["workspace_name", "name"],
    )
 
def downgrade() -> None:
    op.drop_table("dream_definitions")

4.2 DreamType Enum Extension

The DreamType enum in src/schemas/configuration.py is extended:

class DreamType(str, Enum):
    OMNI = "omni"
    TAG = "tag"
    MATERIALIZE = "materialize"
    CONSOLIDATE = "consolidate"

The DreamSettings.ENABLED_TYPES default is updated:

ENABLED_TYPES: list[str] = ["omni"]  # unchanged default; new types opt-in

4.3 Backward Compatibility

  • The default ENABLED_TYPES remains ["omni"]. New dream types must be explicitly enabled.
  • Existing observations without metadata.tags are treated as untagged. The tagging specialist targets these first.
  • The dream_definitions table is new and has no impact on existing data.
  • All new tools are added to new tool lists only (TAGGING_SPECIALIST_TOOLS, CONSOLIDATION_SPECIALIST_TOOLS). Existing DEDUCTION_SPECIALIST_TOOLS and INDUCTION_SPECIALIST_TOOLS are unchanged.

5. Implementation Phases

Phase 1: Foundation (prerequisite: conclusion-tagging spec lands first)

  1. Extend DreamType enum with TAG, MATERIALIZE, CONSOLIDATE.
  2. Add new config fields to DreamSettings: TAGGING_MODEL, CONSOLIDATION_MODEL, tag limits, consolidation limits, CRON_POLL_INTERVAL_SECONDS.
  3. Create Alembic migration for dream_definitions table.
  4. Add DreamDefinition model to src/models.py.
  5. Update DreamPayload with optional definition_id.
  6. Update process_dream in orchestrator.py to route new dream types.

Phase 2: Agent-Generated Tagging

  1. Implement tagging tools in src/utils/agent_tools.py: tag_observations, get_existing_tags, rename_tag, merge_tags.
  2. Implement tool executors for tagging tools in create_tool_executor.
  3. Create TaggingSpecialist in src/dreamer/specialists.py.
  4. Add tag dream type routing in orchestrator.py with run_tagging function.
  5. Enrich observation display in get_recent_observations and search_memory handlers to include tags.
  6. Add optional tag filter parameter to search_memory tool.
  7. Add telemetry event DreamTaggingEvent.

Phase 3: Consolidation Improvements

  1. Implement find_near_duplicates tool and its SQL-based executor.
  2. Implement find_stale_observations tool and executor.
  3. Create ConsolidationSpecialist in src/dreamer/specialists.py.
  4. Add consolidate dream type routing in orchestrator.py.
  5. Implement capacity pressure detection and dynamic prompt injection.
  6. Add telemetry event DreamConsolidationEvent.

Phase 4: Materialized Dreams

  1. Implement CRUD operations for dream_definitions in src/crud/dream_definition.py.
  2. Create API router src/routers/dream_definitions.py with all endpoints.
  3. Add schemas to src/schemas/api.py.
  4. Implement MaterializationSpecialist (dynamic specialist built from definition instructions).
  5. Implement run_materialized_dream in orchestrator.py.
  6. Implement enqueue_materialized_dream in src/deriver/enqueue.py.
  7. Add cron polling to DreamScheduler.
  8. Add on_change trigger hook in the deriver pipeline.
  9. Add telemetry event DreamMaterializeEvent.
  10. Register new router in src/main.py.

Phase 5: Integration and Polish

  1. Update the omni dream type to optionally include tagging as a third phase (controlled by config).
  2. Wire on_change triggers into the deriver observation creation path.
  3. Add croniter dependency to pyproject.toml.
  4. Write comprehensive tests for all new specialists, tools, and API endpoints.
  5. Update CLAUDE.md with new architecture documentation.

6. Files to Modify (exact paths relative to repos/honcho/)

New Files

PathDescription
src/routers/dream_definitions.pyAPI router for dream definition CRUD + trigger
src/crud/dream_definition.pyCRUD operations for dream_definitions table
alembic/versions/xxxx_add_dream_definitions_table.pyAlembic migration for dream_definitions

Modified Files

PathChange
src/models.pyAdd DreamDefinition model
src/schemas/configuration.pyExtend DreamType enum; add ResolvedDreamConfiguration fields
src/schemas/api.pyAdd DreamDefinitionCreate, DreamDefinitionUpdate, DreamDefinitionResponse, DreamTriggerResponse schemas
src/schemas/__init__.pyRe-export new schemas
src/config.pyAdd TAGGING_MODEL, CONSOLIDATION_MODEL, tag limits, consolidation limits, CRON_POLL_INTERVAL_SECONDS to DreamSettings
src/dreamer/specialists.pyAdd TaggingSpecialist, ConsolidationSpecialist; register in SPECIALISTS dict
src/dreamer/orchestrator.pyAdd run_tagging, run_materialized_dream, run_consolidation functions; update process_dream match statement
src/dreamer/dream_scheduler.pyAdd _poll_cron_definitions periodic task; update __init__ to start cron poller
src/utils/agent_tools.pyAdd tag_observations, get_existing_tags, rename_tag, merge_tags, find_near_duplicates, find_stale_observations tools; add TAGGING_SPECIALIST_TOOLS, CONSOLIDATION_SPECIALIST_TOOLS lists; add tag parameter to search_memory; update observation display to include tags
src/utils/queue_payload.pyAdd definition_id to DreamPayload; add create_materialized_dream_payload helper
src/utils/work_unit.pyHandle materialize dream type in work unit key construction (include definition_id)
src/utils/types.pyNo changes needed (DocumentLevel already covers all levels)
src/deriver/enqueue.pyAdd enqueue_materialized_dream function; add on_change trigger check after observation creation
src/main.pyRegister dream_definitions router
src/telemetry/events/dream.pyAdd DreamTaggingEvent, DreamConsolidationEvent, DreamMaterializeEvent
src/telemetry/events/__init__.pyRe-export new events
pyproject.tomlAdd croniter dependency

7. Risk Assessment

RiskSeverityLikelihoodMitigation
Tag vocabulary explosion despite limitsMediumLowHard cap at 50 tags per collection; agent instructed to reuse; merge_tags tool provides cleanup
find_near_duplicates query is expensive on large collectionsHighMediumLimit to collections under CONSOLIDATION_HARD_LIMIT; add query timeout; consider sampling for very large collections; the self-join is bounded by the HNSW index
Cron polling creates excessive dream tasksMediumLowmin_interval_hours prevents spam; LIMIT 10 on the poll query caps throughput; existing dream deduplication index prevents duplicate pending tasks
Materialized dream agent produces low-quality outputMediumMediummax_output_items caps volume; results are stored as regular observations and subject to the same consolidation pressure; developers can disable/delete definitions
Breaking change to DreamPayload with definition_idLowLowField is Optional[str] with None default; existing payloads deserialize without issue
Tagging specialist conflicts with manual metadata writes via APIMediumLowTags live under metadata.tags key; document that this key is managed by the dreamer; API writes to other metadata keys are unaffected
croniter dependency adds supply chain riskLowLowcroniter is a well-maintained, minimal-dependency library widely used in the Python ecosystem

8. Verification Plan

Unit Tests

  1. Tag validation. Test tag format regex, length limits, per-conclusion limits, per-collection limits.
  2. Tag tool executors. Test tag_observations, get_existing_tags, rename_tag, merge_tags against a test database.
  3. Near-duplicate finder. Insert documents with known embeddings at varying cosine distances; verify the query returns correct pairs above threshold.
  4. Staleness scorer. Insert documents with known ages and times_derived values; verify scoring and ordering.
  5. DreamDefinition CRUD. Test create, read, update, delete, and list with filtering.
  6. DreamPayload serialization. Verify definition_id serializes/deserializes correctly; verify backward compatibility with payloads lacking the field.
  7. Work unit key construction. Verify materialize dream type produces correct keys including definition_id.
  8. Cron expression evaluation. Test get_due_cron_definitions with various cron_expression values and last_run_at timestamps.

Integration Tests

  1. Full tagging cycle. Create observations, run TaggingSpecialist, verify tags are applied to metadata.tags.
  2. Full consolidation cycle. Create near-duplicate observations, run ConsolidationSpecialist, verify duplicates are merged/deleted.
  3. Full materialized dream cycle. Create a dream definition, trigger it, verify output observations are created with correct metadata.dream_definition_id.
  4. Omni + tag combined run. Run omni dream with tagging enabled, verify deduction, induction, and tagging all execute.
  5. Cron trigger. Create a cron definition, advance time past the cron interval, verify the cron poller enqueues the dream.
  6. On-change trigger. Create an on_change definition, create new observations via deriver, verify the materialized dream is enqueued.
  7. Capacity pressure. Create observations up to the hard limit, run consolidation, verify the specialist receives pressure directives and reduces count.

API Tests

  1. Dream definitions CRUD endpoints. Full lifecycle: create, get, list, update, delete.
  2. Trigger endpoint. Verify 202 response, verify 429 when under min_interval_hours.
  3. Trigger with disabled definition. Verify 400/422 response.
  4. Conclusion list with tag filter. Verify metadata.tags filtering works in list_conclusions and query_conclusions.

Alembic Tests

  1. Migration up/down. Verify dream_definitions table is created with all constraints and indexes, and cleanly drops on downgrade.
  2. Data preservation. Verify existing documents rows are unaffected by the migration.

9. Open Questions

  1. Should the omni dream type automatically include tagging? Option A: omni always runs deduction + induction + tagging (as phase 3). Option B: omni remains deduction + induction only; tagging is a separate dream type that must be explicitly enabled in ENABLED_TYPES. Current recommendation: Option B for backward compatibility, with a config flag DREAM.OMNI_INCLUDES_TAGGING: bool = False to opt in.

  2. Should tags be visible in the public Conclusion API schema? The metadata field would naturally expose tags. Should we add a dedicated tags field to the Conclusion response schema for convenience, or rely on clients reading metadata.tags? Current recommendation: Expose through metadata only; add convenience field later if demanded.

  3. How should on_change triggers handle high-throughput workspaces? If a workspace creates hundreds of observations per hour, on_change definitions could be triggered very frequently even with min_interval_hours. Should there be an additional rate limit beyond min_interval_hours, such as a minimum observation count threshold? Current recommendation: Add an optional min_observations_since_last_run: int field to dream_definitions (default: 10) that must be met in addition to the time interval.

  4. Should materialized dream results replace previous results or accumulate? When a cron dream runs weekly, should it delete the previous week’s conclusions (tagged with its definition_id) before creating new ones, or let them accumulate? Current recommendation: Add a replace_previous: bool = True field to dream_definitions. When true, soft-delete previous results before creating new ones.

  5. What is the interaction between consolidation and the existing deduction specialist’s deletion capability? The deduction specialist already deletes outdated observations. Should the consolidation specialist be prevented from touching observations created in the current dream cycle? Current recommendation: Yes. The consolidation specialist should have a created_before filter that excludes observations created within the last N hours (configurable, default: 24) to avoid interfering with fresh observations.

  6. Should the find_near_duplicates query use pgvector’s HNSW index or a brute-force self-join? The HNSW index is optimized for point queries (find K nearest neighbors to a single vector), not for finding all pairs above a threshold. A brute-force approach may be necessary for small-to-medium collections. Current recommendation: Use brute-force self-join with a LIMIT and rely on CONSOLIDATION_HARD_LIMIT to keep collection sizes manageable. If performance becomes an issue, batch the query by processing N random seed observations and finding their near-neighbors via HNSW.

  7. Cron expression format — standard 5-field or extended? Standard cron (minute, hour, day-of-month, month, day-of-week) vs. extended cron with seconds. Current recommendation: Standard 5-field cron via croniter, which is simpler and sufficient for the use cases (hourly, daily, weekly).

Graph View

Backlinks