Drain-on-demand: SYNAPBUS_DREAM_PARALLEL (default 1) and
`synapbus memory dream-run --parallel N` fan out N concurrent
dream-agent k8s Jobs per (owner, job_type) in one shot. Set
high (e.g. 8) to drain backlog quickly, then back to 1 for normal
hourly operation.
Schema:
- migration 030_dream_parallelism: adds slot INTEGER NOT NULL DEFAULT 0
to memory_consolidation_jobs. Drops + recreates the partial unique
in-flight index as (owner, job_type, slot) so slots 0..N-1 each hold
one in-flight job independently.
Stores:
- JobsStore.CreateOnSlot + CreateNextAvailableSlot.
- ConsolidatorWorker.ForceRunN dispatches N parallel jobs through the
existing launchOne path (extracted from ForceRun).
- core_rewrite coerces to N=1 regardless of the knob — per-(owner,
agent) blob is wholesale-replace and concurrent rewrites would race.
Three bug fixes discovered while bringing the parallel path up on
kubic:
1. k8s Job names collided on rapid relaunch because runner.go used
"synapbus-<agent>-<msg_id>", and dream dispatches have msg_id=0.
Now appends a unique (timestamp%1e6, 4-byte random) suffix when
msg_id is zero; historical "synapbus-<agent>-<id>" prefix preserved.
2. memory_list_unprocessed didn't actually exclude already-refined
messages — the contract said it should, the implementation
returned the same oldest-50 every cycle. The dream agent kept
re-refining the same set: 221 refines links touched only 55
unique dst messages, so progress flat-lined. Added the
NOT IN (refines/duplicate_of/superseded_by) filter and a
from_agent NOT LIKE 'dream:%' clause so the agent never refines
its own reflections.
3. The k8sjob harness was constructed with nil Waiter in main.go,
so every dream dispatch failed instantly with "k8sjob: no Waiter
configured". Now builds a ClientsetWaiter from the in-cluster
clientset.
Plus admin/server.go gets DreamRunN closure + DefaultDreamParallel
(sourced from MemoryConfig.DreamParallel). admin/socket.go
handleMemoryDreamRun accepts `parallel` arg and returns job_ids[].
CLI admin command grows --parallel N flag.
Live evidence from kubic (image v0.21.0-amd64):
1 CLI call with --parallel 8 produced 8 job rows on slots 0..7,
spawned 8 distinct k8s Jobs with unique suffixes, retired ~86
unprocessed messages in <1 min (vs ~10/cycle for the buggy
serial version pre-fix-2).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Background ConsolidatorWorker dispatches consolidation work to a Claude
Code agent through harness.Harness.Execute (NOT via system DMs — per
feedback_system_dm_no_trigger.md) with a one-time 15m dispatch token.
The dispatched agent uses six new MCP tools, all token-gated and
recording every action to memory_consolidation_jobs.actions JSON.
Stores:
- memory_links.go (+ test): typed edges with actor-prefix reserved-type
guard; AddConsolidationLink bypass for memory_mark_duplicate /
memory_supersede (their contractual writers).
- memory_pins.go (+ test): owner pin overlay, bypasses score floor.
- memory_status.go: queries the memory_status view.
- consolidation_jobs.go: Create / Dispatch / Lease / AppendAction /
Complete with ErrJobAlreadyInFlight via partial unique index.
- auto_links.go: MessageListener generating mention / reply_to /
channel_cooccurrence links automatically on send.
Worker:
- consolidator.go (+ test): ticker pattern modeled on StalemateWorker.
Watermark trigger for link_gen / dedup_contradiction; daily 03:00
UTC for sleep-time core rewrite. Wallclock budget via harness Budget.
Global semaphore gates concurrent owners. Mocked-harness test asserts
no system DM is ever sent.
- consolidator_prompts.go: four job-type prompts passed via env to the
dispatched agent.
MCP tools (internal/mcp/memory_tools.go + test):
- memory_list_unprocessed, memory_write_reflection, memory_rewrite_core,
memory_mark_duplicate, memory_supersede, memory_add_link.
- Full error-code matrix tested per contracts/mcp-memory-tools.md.
- Registered only when SYNAPBUS_DREAM_ENABLED=1.
Injection extensions:
- search/injection.go: pin overlay applied after retrieval; status
filter drops soft_deleted / superseded unless pinned. New
PinProvider, StatusProvider, MessageLookup hooks on InjectionOpts.
Wiring:
- cmd/synapbus/main.go: stores constructed, AutoLinkListener attached
to MessagingService, mcpSrv.SetDream wired, ConsolidatorWorker
start/stop, admin DreamRun closure.
- cmd/synapbus/admin.go: synapbus memory dream-run --owner --job
socket-RPC command (forces a single job bypassing trigger).
Cycle workarounds (documented in code):
- messaging.DreamAgent / HarnessDispatcher are local interfaces (the
agents and harness packages import messaging, not the reverse).
main.go wraps the real types via adapter structs.
Stubbed:
- Cron expression parsing (DreamDeepCron). Hardcoded daily 03:00 UTC.
Adding robfig/cron deferred to keep no-new-deps.
Pre-existing reactor test failures (5) are unchanged; confirmed
pre-020 via stash check.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
- Add `my_status` MCP tool: single call returns agent identity, pending
DMs, channel mentions, system notifications, channel summaries, and
stats with truncation for large inboxes
- Add embeddings CLI: `synapbus embeddings status|reindex|clear` for
managing vectors when switching embedding providers
- Add automatic message retention worker with configurable period
(--message-retention, default 12m), warning notifications 1 month
before deletion, cascade cleanup, and incremental vacuum
- Add manual purge: `synapbus messages purge --older-than --agent --channel`
and `synapbus db vacuum` for on-demand cleanup
- Add `synapbus retention status` CLI for admin visibility
- Create system agent at startup for sending retention warnings
- Filter system agent from discover_agents results
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Move module path from github.com/smart-mcp-proxy/synapbus to
github.com/synapbus/synapbus across all Go imports (47 files).
Add constellation logo options generated via FLUX 1.1 Pro.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Major feature additions across backend and frontend:
Backend:
- Unix domain socket admin server with JSON-RPC protocol
- CLI subcommands: user/agent management, audit, backup, messages, channels
- Managed API keys (sb_ prefix) with permissions, channel limits, expiry
- Thread/reply support in messaging core (reply_to column)
- Context-based trace owner_id propagation for proper audit filtering
- Two-step auth middleware supporting both agent keys and managed API keys
Frontend:
- Complete Slack-like dark theme redesign with custom CSS properties
- Sidebar with Channels, Direct Messages, and Admin sections
- Thread panel (slide-in) for viewing message replies
- API key management page with create form, key display, and
ready-to-use MCP/Claude Code config snippets with copy-to-clipboard
- All pages restyled: login, dashboard, agents, conversations, settings
E2E Tests:
- Fixed test runner binary path resolution
- Added pyproject.toml for test dependencies
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
