synapbus/specs/020-proactive-memory-dream-worker/data-model.md

Data Model — 020-proactive-memory-dream-worker

Reused tables (unchanged)

• messages — every "memory" is a message on a memory-flagged channel. No schema change.
• channels — metadata JSON gains an optional "is_memory": true flag. No schema change.
• agents — owner_id is the canonical scope. No schema change.
• embeddings — existing per-message embeddings used unchanged by search.Service.

New tables (migration 028_memory_consolidation.sql)

memory_core

Per-(owner, agent_name) identity-and-context blob.

CREATE TABLE memory_core (
    owner_id      TEXT NOT NULL,
    agent_name    TEXT NOT NULL,
    blob          TEXT NOT NULL,
    updated_at    DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_by    TEXT NOT NULL,            -- 'human:<owner>' or 'agent:<name>:<token>'
    PRIMARY KEY (owner_id, agent_name)
);
CREATE INDEX idx_memory_core_owner ON memory_core(owner_id);

Constraints:

• LENGTH(blob) <= SYNAPBUS_CORE_MEMORY_MAX_BYTES (default 2048) — enforced in Go, not SQL.
• Replaces wholesale on update — no diff/merge.

memory_links

Directed typed edges between two message IDs.

CREATE TABLE memory_links (
    id             INTEGER PRIMARY KEY AUTOINCREMENT,
    src_message_id INTEGER NOT NULL,
    dst_message_id INTEGER NOT NULL,
    relation_type  TEXT NOT NULL CHECK (relation_type IN (
        'refines', 'contradicts', 'examples', 'related',
        'duplicate_of', 'superseded_by',
        'mention', 'reply_to', 'channel_cooccurrence'
    )),
    owner_id       TEXT NOT NULL,   -- denormalized for fast owner-scoped queries
    created_by     TEXT NOT NULL,   -- 'human:<owner>' / 'agent:<name>:<token>' / 'auto:<rule>'
    metadata       TEXT NOT NULL DEFAULT '{}',
    created_at     DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    UNIQUE(src_message_id, dst_message_id, relation_type)
);
CREATE INDEX idx_memory_links_owner ON memory_links(owner_id);
CREATE INDEX idx_memory_links_src   ON memory_links(src_message_id);
CREATE INDEX idx_memory_links_dst   ON memory_links(dst_message_id);
CREATE INDEX idx_memory_links_type  ON memory_links(relation_type);

Relation types:

• LLM-generated: refines, contradicts, examples, related, duplicate_of, superseded_by
• Auto-generated by messaging layer: mention, reply_to, channel_cooccurrence

memory_consolidation_jobs

Audit log of dispatched dream jobs and their resulting actions.

CREATE TABLE memory_consolidation_jobs (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    owner_id      TEXT NOT NULL,
    job_type      TEXT NOT NULL CHECK (job_type IN (
        'reflection', 'core_rewrite', 'dedup_contradiction', 'link_gen'
    )),
    status        TEXT NOT NULL CHECK (status IN (
        'pending', 'dispatched', 'running', 'succeeded', 'partial', 'failed', 'expired'
    )) DEFAULT 'pending',
    trigger_reason TEXT NOT NULL,            -- 'watermark:N', 'cron:nightly', 'manual:<owner>'
    dispatch_token TEXT,                     -- FK-ish into memory_dispatch_tokens.token
    harness_run_id TEXT,                     -- FK into harness_runs.id (existing)
    actions       TEXT NOT NULL DEFAULT '[]', -- JSON array of {tool, args, before, after}
    summary       TEXT,                       -- human-readable summary written on completion
    error         TEXT,
    lease_until   DATETIME,                   -- in-flight lease; null when not running
    started_at    DATETIME,
    finished_at   DATETIME,
    created_at    DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_consolidation_owner_status ON memory_consolidation_jobs(owner_id, status);
CREATE INDEX idx_consolidation_lease        ON memory_consolidation_jobs(lease_until)
    WHERE status = 'running';
CREATE UNIQUE INDEX idx_consolidation_in_flight
    ON memory_consolidation_jobs(owner_id, job_type)
    WHERE status IN ('pending', 'dispatched', 'running');

The partial-unique index enforces: at most one in-flight job per (owner, job_type).

memory_pins

Owner-pinned message IDs that bypass the relevance floor.

CREATE TABLE memory_pins (
    owner_id   TEXT NOT NULL,
    message_id INTEGER NOT NULL,
    pinned_by  TEXT NOT NULL,           -- 'human:<owner>'
    note       TEXT,                     -- why this is pinned (free-form)
    pinned_at  DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (owner_id, message_id)
);
CREATE INDEX idx_memory_pins_owner ON memory_pins(owner_id);

memory_dispatch_tokens

Single-use, owner-bound, job-bound tokens.

CREATE TABLE memory_dispatch_tokens (
    token              TEXT PRIMARY KEY,    -- 32-byte random, base64url
    owner_id           TEXT NOT NULL,
    consolidation_job_id INTEGER NOT NULL,
    issued_at          DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    expires_at         DATETIME NOT NULL,    -- issued_at + 15m
    used_at            DATETIME,
    revoked_at         DATETIME,
    FOREIGN KEY (consolidation_job_id) REFERENCES memory_consolidation_jobs(id)
);
CREATE INDEX idx_memory_tokens_job ON memory_dispatch_tokens(consolidation_job_id);

Token is valid when revoked_at IS NULL AND expires_at > now() AND consolidation_job_id == claimed job. used_at becomes informational once first set; subsequent calls within the same job are allowed.

memory_injections

24-hour rolling ring of what was injected for each tool call.

CREATE TABLE memory_injections (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    owner_id     TEXT NOT NULL,
    agent_name   TEXT NOT NULL,
    tool_name    TEXT NOT NULL,
    packet_size_chars INTEGER NOT NULL,
    packet_items_count INTEGER NOT NULL,
    message_ids  TEXT NOT NULL,        -- JSON array of message_ids included
    core_blob_included BOOLEAN NOT NULL DEFAULT 0,
    created_at   DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_memory_injections_owner_time ON memory_injections(owner_id, created_at);

A daily cleanup query (run by the consolidator worker) deletes created_at < now() - 24h.

View: memory_status

Derived from memory_consolidation_jobs.actions. Tells retrieval whether a message is active / soft-deleted / superseded.

CREATE VIEW memory_status AS
WITH actions AS (
    SELECT
        owner_id,
        json_extract(act.value, '$.target_message_id')  AS message_id,
        json_extract(act.value, '$.tool')                AS tool,
        json_extract(act.value, '$.args.keep_id')        AS keep_id,
        json_extract(act.value, '$.args.b_id')           AS superseded_by,
        json_extract(act.value, '$.args.reason')         AS reason,
        finished_at                                       AS at
    FROM memory_consolidation_jobs j, json_each(j.actions) act
    WHERE j.status IN ('succeeded', 'partial')
)
SELECT
    message_id,
    owner_id,
    CASE
        WHEN MAX(CASE WHEN tool='memory_supersede' THEN at END) IS NOT NULL THEN 'superseded'
        WHEN MAX(CASE WHEN tool='memory_mark_duplicate' AND keep_id != message_id THEN at END) IS NOT NULL THEN 'soft_deleted'
        ELSE 'active'
    END AS status,
    MAX(CASE WHEN tool='memory_supersede' THEN superseded_by END) AS superseded_by,
    MAX(CASE WHEN tool='memory_mark_duplicate' AND keep_id != message_id THEN at END) AS soft_deleted_at,
    MAX(reason) AS reason
FROM actions
GROUP BY message_id, owner_id;

Retrieval excludes status != 'active' unless pinned or include_inactive=true.

State transitions

Memory message (active by default — not in memory_status)
   │
   ├── memory_mark_duplicate(keep != self) → soft_deleted   (via view)
   │       └── owner restore action overrides              → active
   │
   ├── memory_supersede(b)                 → superseded     (via view)
   │       └── owner restore                                → active
   │
   └── (no consolidation event)            → active

Pinning and protection are orthogonal flags (in memory_pins; protection is currently piggybacked on a protected_until JSON field in memory_pins.note — promoted to a column in Phase 2 if needed).

Indexes & query patterns

Hot queries:

1. Injection retrieval: search.Service.Search() already paginates → join memory_status to drop non-active → drop in pin overlay → apply token budget.
2. Dream worker watermark check: SELECT COUNT(*) FROM messages m JOIN agents a ON m.from_agent=a.name WHERE a.owner_id=? AND m.channel_id IN (memory_channels) AND m.id > last_reflection_max_id. Owner-filtered + range — uses existing message indexes.
3. Audit lookup: SELECT * FROM memory_consolidation_jobs WHERE owner_id=? ORDER BY created_at DESC LIMIT 50. Covered by idx_consolidation_owner_status.