Dream phase

Status: read this once before scheduling anything in prod. Phase-1 audits run read-only (no side effects). Phase-2 planners plan in dry-run and, in apply mode, mutate the relational store through bi-temporal soft-delete and in-place updates. The graph-store mirror for the vectorcypher mutation ops is deferred to a future release. Guardrails protect the apply path: the KHORA_DREAM_DISABLE_APPLY env-var kill-switch, an advisory lock held through apply, a chunk_id runtime assertion, and snapshot-before-mutate undo records. Default config is enabled=False. Nothing happens until you opt in.

​

When to use it

• You suspect duplicate entities from independent ingest batches ("OpenAI" and "Open AI", "Marie Curie" from two different documents).
• Recall latency is creeping up because chunks.source_chunk_ids arrays carry references to long-gone chunks.
• An ExpertiseConfig change landed weeks ago and you don’t know whether the data has caught up.
• Operators want a periodic, human-readable “state of the graph” snapshot they can review before authorizing destructive ops.

​

Quickstart

from khora import Khora, KhoraConfig, DreamConfig

kb = Khora(
    KhoraConfig(
        dream=DreamConfig(
            enabled=True,
            report_file_sink_enabled=True,
            report_event_sink_enabled=False,
            report_collector_sink_enabled=True,
        ),
    ),
)

result = await kb.dream(namespace_id, mode="dry-run")

# `result.ops` is the aggregate counter - one OpSummary per op kind.
for summary in result.ops:
    print(
        f"{summary.op_type}: planned={summary.planned} "
        f"skipped={summary.skipped} failed={summary.failed}"
    )

# Per-op detail (decision strings, outputs dicts, rationale) lives in
# the file sink's events.jsonl, or in `result.metadata["plan_payload"]`
# in dry-run mode.

​

Operations

​

Audit operations (Phase 1)

​

VectorCypher: schema drift vs ExpertiseConfig

​

VectorCypher: PageRank-based orphan report

• PR score in the bottom orphan_pr_percentile_threshold percentile (default 5)
• mention_count ≤ 1
• No recent recall hits

​

VectorCypher: source_chunk_ids array-length audit

​

Planner operations (Phase 2)

​

VectorCypher: cross-batch entity dedupe

• Tighter than the online resolver’s 0.85 default. This is the cross-batch pass with the benefit of all accumulated evidence.
• Skip-collisions (the same canonical entity would absorb two clusters) are reported under outputs.skip_collision_count. The op doesn’t choose for you.
• See: Köpcke & Rahm 2010 on entity-matching frameworks.

​

VectorCypher: centroid recompute

​

VectorCypher: source_chunk_ids GC

​

Output channels (sinks)

​

File sink

​

Event sink

​

Collector sink (OpenTelemetry)

• khora.dream.run, khora.dream.phase, khora.dream.llm_call (reserved), khora.dream.undo (reserved)

• khora.dream.vectorcypher.{schema_drift,orphan_report,source_chunk_ids_audit,source_chunk_ids_gc,centroid_recompute,dedupe_entities,prune_edges,normalize_schema,contradiction_detect,community_summary}

• khora.dream.runs_total {trigger, outcome}
• khora.dream.run.duration {trigger, outcome} (histogram, seconds)
• khora.dream.phase.duration {phase, outcome} (histogram)
• khora.dream.ops_total {phase, op_type, decision}
• khora.dream.llm.tokens {direction, model} (reserved)
• khora.dream.undo_invocations_total {op_type, outcome} (reserved)
• khora.dream.report.write_failures_total {reason} (internal)

​

API reference

​

Top-level entry points

async def dream(
    namespace: str | UUID,
    *,
    mode: str = "dry-run",                # "dry-run" | "apply"
    scope: DreamScope | None = None,
    ops: Iterable[OpKind] | None = None,
    config: DreamConfig | None = None,
    on_progress: Callable[[DreamProgress], None] | None = None,
    resume_from: UUID | None = None,
) -> DreamResult

async def dream_status(run_id: UUID) -> dict[str, object]
async def dream_history(namespace: str | UUID, *, limit: int = 20) -> list[DreamRunInfo]

​

Configuration

​

Op kinds

​

Postgres-only apply ops

• vectorcypher_dedupe_entities
• vectorcypher_centroid_recompute
• vectorcypher_prune_edges
• vectorcypher_source_chunk_ids_gc

​

Plan / scope / result dataclasses

• DreamScope(op_kinds, since, until, entity_ids, document_ids): public. None fields = no restriction.
• DreamResult(run, diff, ops, llm_usage, metadata): public. metadata carries plan_hash + plan_payload on dry-run.
• DreamRunInfo(run_id, namespace_id, mode, started_at, finished_at, duration_ms, resume_of): public.
• DreamMode = Literal["dry-run", "apply"]: public.
• UndoRecord(op_id, op_type, before, applied_at): public. Returned by every apply handler, persisted into undo.json (schema dream-undo/1).
• OpSummary(op_type, planned, applied, skipped, failed): public. The shape of items in DreamResult.ops.
• DreamOp, DreamPlan, Checkpoint, DreamDiff, DreamProgress: internal.

​

Planner functions

# Vectorcypher (khora.dream.engines.vectorcypher)
plan_vectorcypher_schema_drift(namespace_id, *, coordinator, expertise, previous_run_id=None, ...)
plan_vectorcypher_orphan_report(namespace_id, *, coordinator, expertise=None, pr_percentile_threshold=5.0, ...)
plan_vectorcypher_source_chunk_ids_audit(namespace_id, *, coordinator, top_k_offenders=20)
plan_vectorcypher_source_chunk_ids_gc(namespace_id, *, coordinator, min_dead=1)
plan_vectorcypher_centroid_recompute(namespace_id, *, coordinator, merge_clusters, ...)
plan_vectorcypher_dedupe_entities(namespace_id, *, coordinator, default_threshold=0.90, ...)
plan_vectorcypher_normalize_schema(namespace_id, *, coordinator, config) -> list[DreamOp]

​

Apply functions

# Vectorcypher
apply_vectorcypher_source_chunk_ids_gc(op, *, coordinator, session) -> UndoRecord
apply_vectorcypher_centroid_recompute(op, *, coordinator, session, embedder=None) -> UndoRecord
apply_vectorcypher_dedupe_entities(op, *, coordinator, session) -> UndoRecord
apply_vectorcypher_normalize_schema(op, *, coordinator, session) -> UndoRecord

​

Reporting sinks

class DreamFileSink(base_dir: str | Path, *, redact_text: Literal["none","summary","all"] = "summary")
class DreamEventSink(dispatcher: HookDispatcher, *, delivery: Literal["sync","async"] = "async",
                     outbox_maxsize: int = 10_000, subscription_class: str = "dream")
class DreamCollectorSink()  # stateless; OTel spans + metrics

​

Exceptions

• DreamDisabledError: DreamConfig.enabled is False
• DreamApplyDisabled: KHORA_DREAM_DISABLE_APPLY env-var kill-switch tripped, mode="apply" refused without touching the DB
• DreamForbiddenOpError: plan contains a forbidden op (document delete, chunk_id mutation, UNIQUE-collision write, read-only namespace, sub-floor retention)
• DreamRunStuckError(run_id, heartbeat_age_seconds): prior run is applying with a stale heartbeat. Resolve via resume_from=run_id after manual review
• DreamLockUnavailable(namespace_id, timeout_seconds): advisory lock contention

​

Protocol

@runtime_checkable
class DreamCapable(Protocol):
    @property
    def dream_capabilities(self) -> frozenset[OpKind]: ...
    async def plan_dream(self, namespace_id, *, scope, config, expertise=None) -> DreamPlan: ...
    async def apply_dream(self, plan, *, checkpoint=None, on_progress=None) -> DreamResult: ...

​

Storage substrate

​

Checkpoint table: khora_dream_runs

​

Bi-temporal columns

​

Concurrency

​

Research & Prior Art

​

Sleep and memory consolidation in neuroscience

​

Offline RL replay buffers

​

Agentic memory frameworks

​

Knowledge-graph maintenance

​

Tombstones, soft-delete, and bi-temporal modeling

​

OLTP vs. OLAP

​

Honest limits

• Decide when to run. That’s operator policy: cron / Temporal / k8s CronJob.
• Validate that a planned op makes business sense. The planner uses heuristics (cosine, Levenshtein, age thresholds). Operators are expected to dry-run several times and review the file-sink reports before flipping to apply.
• Reverse an applied op automatically. Undo records are persisted to undo.json (schema dream-undo/1) but there is no kb.undo(run_id) API. Restoring from the snapshot is a hand-rolled operation. An automated undo player is planned.
• Replace good ingest-time decisions. If dedupe finds 10,000 candidate merges in a fresh namespace, the bug is upstream (in the extraction pipeline, the embedding model choice, or the per-type thresholds), not in dream phase.

​

Stability

​

Not yet implemented

• Phase 5: advanced operations. Community detection + LLM-generated summaries (GraphRAG-style, the first dream-phase ops to actually call an LLM, gated by the existing llm_max_tokens_per_run budget), edge pruning by weight × recency, contradiction detection across memory_facts, schema-drift normalization with an operator-supplied mapping.
• apply_vectorcypher_centroid_recompute on SQLite-LanceDB. The handler is Postgres-only because the embedded vector backend’s update_entity_embedding commits out-of-band, violating the caller-owned-transaction contract. The SQLite path needs a session-aware vector-backend write API first.
• apply_vectorcypher_dedupe_entities Neo4j re-target. The Postgres relationships rewrite ships today. The Neo4j edge re-target + archived-node path needs a different transactional shape (no shared session with PG) and is deferred.
• Auto-chaining of dedupe → centroid_recompute. Today centroid_recompute in the default plan dispatch receives merge_clusters=[] and emits no DreamOps. Real centroid plans require direct invocation with clusters from a prior dedupe run, or manual operator stitching via resume_from.
• Planner mode="apply" cleanup. Two of the Phase 2 planners (plan_vectorcypher_dedupe_entities, plan_vectorcypher_source_chunk_ids_gc, plan_vectorcypher_centroid_recompute) still accept a mode="apply" kwarg that raises NotImplementedError. The orchestrator never sets this kwarg. The dedicated apply_<op> functions are the real apply entry point. The raise path is dead code reachable only from direct callers; removing the kwarg is a follow-up cleanup.