From ea397869976c154e3917b9cefbf8b0ed97e3eee1 Mon Sep 17 00:00:00 2001 From: Jamie Pine Date: Mon, 2 Mar 2026 03:53:15 -0800 Subject: [PATCH 1/5] =?UTF-8?q?feat:=20cortex=20topic=20synthesis=20?= =?UTF-8?q?=E2=80=94=20targeted=20context=20documents=20replacing=20global?= =?UTF-8?q?=20bulletin?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds living context documents (topics) that are addressable, per-channel, injectable into workers, and maintained by the cortex sync loop. Data layer: topics + topic_versions tables, TopicStore with full CRUD, versioning, channel assignment, and staleness tracking. Synthesis: topic sync loop runs alongside existing cortex loops, gathers memories matching configurable criteria (query, types, importance, pinned memories), synthesizes via LLM, stores versioned snapshots. Stale detection based on newest matching memory timestamp. Delivery: channels compose channel-specific topics (falling back to global bulletin), workers accept topic_ids param for context injection, cortex chat uses all active topics, task workers read topic_ids from task metadata. API: full CRUD + sync trigger + version history endpoints. Interface: topics tab with card grid browser, create dialog with criteria form + channel assignments, detail dialog with content/config/versions tabs, pin memory search, manual sync trigger. --- docs/design-docs/cortex-topic-synthesis.md | 426 ++++++++ interface/src/api/client.ts | 125 +++ interface/src/components/AgentTabs.tsx | 1 + interface/src/router.tsx | 18 + interface/src/routes/AgentTopics.tsx | 1028 ++++++++++++++++++++ migrations/20260302000001_topics.sql | 31 + prompts/en/cortex_topic_synthesis.md.j2 | 21 + src/agent/channel.rs | 63 +- src/agent/channel_dispatch.rs | 20 +- src/agent/cortex.rs | 260 +++++ src/agent/cortex_chat.rs | 22 +- src/api.rs | 1 + src/api/agents.rs | 11 + src/api/server.rs | 14 +- src/api/state.rs | 8 + src/api/topics.rs | 296 ++++++ src/config.rs | 10 + src/lib.rs | 2 + src/main.rs | 11 + src/memory/search.rs | 6 +- src/prompts/engine.rs | 10 + src/tools/spawn_worker.rs | 10 + src/topics.rs | 5 + src/topics/store.rs | 605 ++++++++++++ tests/bulletin.rs | 1 + tests/context_dump.rs | 1 + 26 files changed, 2997 insertions(+), 9 deletions(-) create mode 100644 docs/design-docs/cortex-topic-synthesis.md create mode 100644 interface/src/routes/AgentTopics.tsx create mode 100644 migrations/20260302000001_topics.sql create mode 100644 prompts/en/cortex_topic_synthesis.md.j2 create mode 100644 src/api/topics.rs create mode 100644 src/topics.rs create mode 100644 src/topics/store.rs diff --git a/docs/design-docs/cortex-topic-synthesis.md b/docs/design-docs/cortex-topic-synthesis.md new file mode 100644 index 000000000..48607c616 --- /dev/null +++ b/docs/design-docs/cortex-topic-synthesis.md @@ -0,0 +1,426 @@ +# Cortex Topic Synthesis + +Replace the single-bulletin broadcast with targeted, living context documents. Each topic is an addressable unit of knowledge — synthesized from a memory subset, kept fresh by the cortex, and injectable into any process by ID. + +## Why + +The bulletin is a global broadcast. Every channel and every worker spawned from a channel gets the same ~1500 word context block regardless of what they're actually working on. A Discord moderation channel gets the same context as a coding channel. A worker spun up to refactor the auth module gets the same context as one checking the weather. + +Topics fix this by making context **addressable** and **targeted**: + +- A channel gets topics assigned to it — only the context it needs. +- A worker gets spawned with `topic_ids` — instant deep context on the specific project it's about to work on. +- The cortex maintains each topic independently — different refresh rates, different memory criteria, different scope. +- Humans can curate topics directly — pin specific memories, edit the synthesis, adjust criteria. + +The bulletin becomes just another topic (a broad "General Context" topic) rather than a special system. + +## What Exists Today + +**The bulletin system** (`src/agent/cortex.rs`): +- `spawn_bulletin_loop()` runs on a timer (default 1 hour) +- `gather_bulletin_sections()` queries 8 hardcoded memory categories (Identity, Recent, Decisions, Important, Preferences, Goals, Events, Observations) plus active tasks +- `generate_bulletin()` feeds the raw sections through an LLM to synthesize a ~1500 word briefing +- Result is stored in `RuntimeConfig::memory_bulletin` as `ArcSwap` +- Every channel reads the same bulletin via `rc.memory_bulletin.load()` in `build_system_prompt()` +- Workers receive no bulletin context — they only get the task description + +**The `channels.bulletin` column** exists in the schema but is never written to. It was an early hook for per-channel bulletins. + +**The signal buffer** in `Cortex` already collects `MemorySaved` signals from the event bus, which can drive topic staleness detection. + +## Data Model + +### `topics` table + +```sql +CREATE TABLE IF NOT EXISTS topics ( + id TEXT PRIMARY KEY, + agent_id TEXT NOT NULL, + title TEXT NOT NULL, + content TEXT NOT NULL DEFAULT '', + criteria TEXT NOT NULL, -- JSON: TopicCriteria + pin_ids TEXT NOT NULL DEFAULT '[]', -- JSON: array of memory IDs always included + channel_ids TEXT NOT NULL DEFAULT '[]', -- JSON: array of channel_ids this routes to + status TEXT NOT NULL DEFAULT 'active', + max_words INTEGER NOT NULL DEFAULT 1500, + last_memory_at TEXT, -- newest memory timestamp used in last sync + last_synced_at TEXT, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')) +); + +CREATE INDEX IF NOT EXISTS idx_topics_agent ON topics(agent_id); +CREATE INDEX IF NOT EXISTS idx_topics_status ON topics(status); +``` + +### `topic_versions` table + +```sql +CREATE TABLE IF NOT EXISTS topic_versions ( + id TEXT PRIMARY KEY, + topic_id TEXT NOT NULL, + content TEXT NOT NULL, + memory_count INTEGER NOT NULL, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + FOREIGN KEY (topic_id) REFERENCES topics(id) ON DELETE CASCADE +); + +CREATE INDEX IF NOT EXISTS idx_topic_versions_topic ON topic_versions(topic_id); +``` + +### Topic status + +``` +active — cortex syncs this topic on schedule +paused — exists but skipped during sync +archived — hidden from UI, never synced, kept for history +``` + +### `TopicCriteria` (JSON in `criteria` column) + +```rust +struct TopicCriteria { + /// Semantic query for hybrid search (optional — omit for pure metadata queries) + query: Option, + /// Filter to specific memory types (empty = all types) + memory_types: Vec, + /// Minimum importance threshold + min_importance: Option, + /// Only memories from these channels (empty = all channels) + channel_ids: Vec, + /// Only memories newer than this duration ago (e.g. "30d", "7d") + max_age: Option, + /// Search mode override (defaults to Hybrid if query present, Recent otherwise) + mode: Option, + /// Sort order for non-hybrid modes + sort_by: Option, + /// Max memories to gather before synthesis (default 30) + max_memories: usize, +} +``` + +This maps directly to the existing `SearchConfig` plus a few SQL-level filters (`channel_ids`, `max_age`) that need minor additions to `MemoryStore::get_sorted()`. No new query language — just structured parameters over existing search infrastructure. + +## Rust Types + +``` +src/topics/mod.rs — Topic, TopicStatus, TopicCriteria, TopicVersion types + re-exports +src/topics/store.rs — TopicStore CRUD (SQLite) +``` + +### TopicStore API + +```rust +impl TopicStore { + pub async fn create(&self, topic: &Topic) -> Result<()>; + pub async fn get(&self, id: &str) -> Result>; + pub async fn list(&self, agent_id: &str) -> Result>; + pub async fn list_active(&self, agent_id: &str) -> Result>; + pub async fn update(&self, topic: &Topic) -> Result<()>; + pub async fn delete(&self, id: &str) -> Result<()>; + pub async fn get_for_channel(&self, agent_id: &str, channel_id: &str) -> Result>; + pub async fn save_version(&self, version: &TopicVersion) -> Result<()>; + pub async fn get_versions(&self, topic_id: &str, limit: i64) -> Result>; +} +``` + +`TopicStore` goes into `AgentDeps` alongside `MemorySearch`, `TaskStore`, etc. + +## Synthesis + +### Topic sync loop + +New cortex loop: `spawn_topic_sync_loop()`. Runs alongside the existing bulletin and association loops. + +``` +spawn_topic_sync_loop(deps, logger) + │ + └─ Startup: sync all active topics (same as bulletin does initial generation) + │ + └─ Every topic_sync_interval_secs (default 600): + │ + ├─ Load all active topics for this agent + │ + └─ For each topic: + ├─ Staleness check: + │ Query newest memory matching criteria + │ Stale if newest_memory.created_at > topic.last_memory_at + │ Stale if topic.last_synced_at is None (never synced) + │ + ├─ If not stale → skip + │ + └─ If stale → synthesize: + ├─ gather_topic_memories(criteria, pin_ids) + │ Uses existing MemorySearch with SearchConfig built from criteria + │ Loads pinned memories by ID, prepends to results + │ Deduplicates + │ + ├─ Format as raw sections (markdown, same style as bulletin) + │ + ├─ LLM synthesis via topic_synthesis.md.j2 + │ System: "Synthesize a topic document about: {title}" + │ User: "Synthesize into {max_words} words:\n{raw_sections}" + │ + ├─ Update topic.content, last_synced_at, last_memory_at + │ + ├─ Save TopicVersion snapshot + │ + └─ Log cortex_event: topic_synced +``` + +The LLM call per topic is the same weight as the bulletin call — one synthesis pass, no tools, branch-tier model. With 5-10 active topics on a 10-minute interval, that's 5-10 cheap LLM calls every 10 minutes, but only for stale topics. In practice most ticks sync 0-2 topics. + +### Prompt template + +New file: `prompts/en/cortex_topic_synthesis.md.j2` + +System prompt tells the LLM to synthesize a focused context document about the topic title. Same constraints as the bulletin synthesis (no raw IDs, no section headers from input, prioritize recent and high-importance, keep it scannable) but scoped to the topic. + +The synthesis fragment can reuse `fragments/system/cortex_synthesis.md.j2` since it's the same structure — max_words + raw_sections. + +## Delivery + +### Workers (the primary use case) + +Workers spawned with topic IDs get the synthesized content injected above their task prompt. + +**Changes to `SpawnWorkerArgs`:** +```rust +pub struct SpawnWorkerArgs { + pub task: String, + pub interactive: bool, + pub suggested_skills: Vec, + pub worker_type: Option, + pub directory: Option, + pub topic_ids: Vec, // new +} +``` + +**In `spawn_worker_from_state()`:** + +After building the worker task with temporal context, before creating the Worker: + +```rust +if !topic_ids.is_empty() { + let mut topic_context = String::from("## Topic Context\n\n"); + for id in &topic_ids { + if let Ok(Some(topic)) = topic_store.get(id).await { + topic_context.push_str(&format!("### {}\n\n{}\n\n", topic.title, topic.content)); + } + } + worker_task = format!("{topic_context}---\n\n{worker_task}"); +} +``` + +Topic content goes into the task message (the first user message), not the system prompt. The worker template already says "Your task is provided in the first message. It contains everything you need to know." Topics are just more of "everything you need to know." + +The same injection applies to cortex-spawned task workers in `pickup_one_ready_task()`. Task metadata can carry `topic_ids`: + +```json +{ + "topic_ids": ["uuid-1", "uuid-2"], + "worker_type": "opencode", + "directory": "/path/to/project" +} +``` + +### Channels + +Channels with topic assignments receive composed topic content instead of (or alongside) the global bulletin. + +**In `build_system_prompt()`:** + +```rust +let channel_topics = topic_store.get_for_channel(&agent_id, &channel_id).await?; +let memory_context = if channel_topics.is_empty() { + // Fall back to global bulletin + rc.memory_bulletin.load().to_string() +} else { + channel_topics.iter() + .map(|t| format!("### {}\n\n{}", t.title, t.content)) + .collect::>() + .join("\n\n") +}; +``` + +The template slot stays the same — `{{ memory_bulletin }}` in `channel.md.j2` just receives different content. No template changes needed. + +Channels with no topic assignments continue to receive the global bulletin. The bulletin loop doesn't need to change or stop — it serves as the default. + +### Cortex chat + +Cortex chat currently receives the global bulletin. It should receive all active topics composed together, since the cortex chat is the system-wide reasoning interface. + +## Topic-Aware Spawning (LLM Side) + +The channel's `spawn_worker` tool description gets updated to include `topic_ids` as an optional parameter. The channel prompt should mention available topics so the LLM can reference them when spawning workers. + +**In the status block or a new section:** +``` +## Available Topics +- "Platform Refactor" (id: abc123) — last synced 5m ago +- "Auth System" (id: def456) — last synced 12m ago +``` + +This lets the channel LLM make informed decisions about which topics to attach when delegating work. + +## API Endpoints + +``` +GET /api/agents/topics — list all topics for agent +POST /api/agents/topics — create topic +GET /api/agents/topics/:id — get topic with current content +PUT /api/agents/topics/:id — update topic (title, criteria, channels, status, etc.) +DELETE /api/agents/topics/:id — delete topic +POST /api/agents/topics/:id/sync — trigger manual re-sync +GET /api/agents/topics/:id/versions — get version history +``` + +Handler file: `src/api/topics.rs`. Follows the same patterns as `src/api/tasks.rs` and `src/api/cron.rs`. + +## Interface + +New tab: **Topics** (between Memories and Workers in the tab order). + +### Topic browser + +Card grid showing all topics. Each card displays: +- Title +- Status badge (active/paused/archived) +- Last synced timestamp (relative, e.g. "5m ago") +- Channel count (how many channels assigned) +- Word count of current content +- Memory count from last sync + +### Topic detail / editor + +Two-column layout: + +**Left — Content:** +- Rendered markdown of the current synthesis +- Word count +- "Last synced" timestamp with manual sync button + +**Right — Configuration:** +- Title (editable) +- Status toggle (active/paused/archived) +- Criteria editor — form fields mapping to `TopicCriteria`: + - Semantic query (text input) + - Memory types (multi-select pills) + - Min importance (slider, 0.0-1.0) + - Channel filter (multi-select from known channels) + - Max age (dropdown: 7d, 30d, 90d, all) + - Max memories (number input) +- Channel assignments (checkbox list of all channels) +- Pinned memories (search + add interface — search memories, click to pin) + +### Version history / diff + +Below the content panel: a version timeline. Click any two versions to see a diff. Standard inline diff rendering (insertions green, deletions red). Versions are timestamped and show memory count. + +### Topic creation + +"New Topic" button opens a form with title, criteria, and channel assignments. Content starts empty — first sync populates it. + +## Config + +```rust +pub struct CortexConfig { + // ... existing fields ... + + /// Interval between topic sync passes (default: 600s = 10 min) + pub topic_sync_interval_secs: u64, +} +``` + +Per-topic `max_words` is stored on the topic itself (default 1500), not in the global config. + +## Bulletin Deprecation Path + +The bulletin doesn't need to die immediately. + +1. **Coexistence.** Topics exist alongside the bulletin. Channels with topic assignments use topics. Channels without use the bulletin. Workers optionally receive topics. + +2. **Bulletin becomes a topic.** Once topics are working, create a "General Context" topic with broad criteria (the same 8 categories the bulletin currently uses). Assign it to channels that don't need specific topics. The bulletin loop continues to run in parallel for backward compatibility. + +3. **Bulletin loop removed.** Once all channels have topic assignments and the "General Context" topic is proven stable, the bulletin loop can be removed. `RuntimeConfig::memory_bulletin` stays as a read path that loads the "General Context" topic content. + +No rush on steps 2 and 3. The bulletin works fine as a default. + +## Phases + +### Phase 1: Data Layer + +- Migration for `topics` and `topic_versions` tables +- `Topic`, `TopicVersion`, `TopicStatus`, `TopicCriteria` types +- `TopicStore` with full CRUD + version history +- Add `TopicStore` to `AgentDeps` +- Add `channel_ids` and `max_age` filter support to `MemoryStore::get_sorted()` (minor SQL additions) + +### Phase 2: Synthesis + Sync Loop + +- `gather_topic_memories()` — build `SearchConfig` from `TopicCriteria`, query, merge pinned, dedup +- `synthesize_topic()` — format raw sections, LLM synthesis, save content + version +- `spawn_topic_sync_loop()` — startup sync + periodic staleness check +- `cortex_topic_synthesis.md.j2` prompt template +- Cortex event logging for `topic_synced` / `topic_sync_failed` + +### Phase 3: Delivery — Workers + +- Add `topic_ids` to `SpawnWorkerArgs` and the `spawn_worker` tool JSON schema +- Inject topic content in `spawn_worker_from_state()` +- Inject topic content in `pickup_one_ready_task()` (read from task metadata) +- Update `spawn_worker` tool description to mention topic IDs + +### Phase 4: Delivery — Channels + +- `get_for_channel()` query in topic store +- Modify `build_system_prompt()` to compose channel topics, fall back to bulletin +- Inject available topics list into status block or a new prompt section +- Update cortex chat to receive all active topics + +### Phase 5: API + +- `src/api/topics.rs` handlers (list, create, get, update, delete, sync, versions) +- Register routes in `src/api/server.rs` +- TypeScript client types and methods in `interface/src/api/client.ts` + +### Phase 6: Interface + +- Topic browser page (`AgentTopics.tsx`) +- Topic detail/editor with criteria form and channel assignments +- Pin memory search UI +- Version history with diff view +- Tab in `AgentTabs.tsx` +- Route in `router.tsx` + +### Phase 7: Signal-Driven Refresh + +- On `MemorySaved` signal in cortex, evaluate which active topics match the new memory +- Mark matching topics as stale (bump priority in next sync pass) +- Optionally trigger immediate sync for topics whose channels are currently active + +## Phase Ordering + +``` +Phase 1 (data) — standalone +Phase 2 (synthesis) — depends on Phase 1 +Phase 3 (workers) — depends on Phase 1 (reads from store), independent of Phase 2 (content may be empty) +Phase 4 (channels) — depends on Phase 1, benefits from Phase 2 +Phase 5 (API) — depends on Phase 1 +Phase 6 (interface) — depends on Phase 5 +Phase 7 (signals) — depends on Phase 2 +``` + +Phases 2, 3, 4, and 5 can run in parallel after Phase 1. Phase 6 needs Phase 5. Phase 7 needs Phase 2. + +## Open Questions + +**Topic creation by cortex.** Should the cortex be able to create topics automatically? For example, when it detects a recurring cluster of related memories across channels, it could propose a topic with pre-filled criteria. This mirrors how cortex promotes todos into tasks — it could promote memory patterns into topics. Probably a Phase 7+ feature, after the manual creation path is proven. + +**Topic size budget.** Multiple topics assigned to a channel all consume context window space. If a channel has 5 topics at 1500 words each, that's 7500 words of topic context — potentially crowding out conversation history. Options: enforce a per-channel total word budget, let the channel LLM request specific topics instead of receiving all assigned ones, or trust the operator to assign topics sensibly. Starting with trust and adding a budget later if it becomes a problem. + +**Topic sharing across agents.** In multi-agent setups, should topics be shareable? A "Codebase Architecture" topic might be useful to multiple agents. The current design scopes topics to `agent_id`. Cross-agent sharing could be added later by introducing a `shared` flag or a separate `org_topics` table. + +**Memory pinning UX.** The pin_ids field lets humans force specific memories into a topic regardless of criteria match. The UI needs a way to search memories and add them as pins. This should use the existing memory search API (`/api/agents/memories/search`) with a "pin to topic" action. diff --git a/interface/src/api/client.ts b/interface/src/api/client.ts index 482d056d1..9c8bc285e 100644 --- a/interface/src/api/client.ts +++ b/interface/src/api/client.ts @@ -945,6 +945,83 @@ export interface UpdateTaskRequest { approved_by?: string; } +// -- Topic Types -- + +export type TopicStatus = "active" | "paused" | "archived"; +export type SearchMode = "hybrid" | "vector" | "fulltext" | "typed" | "recent"; +export type SearchSort = "recent" | "importance" | "most_accessed"; + +export interface TopicCriteria { + query?: string; + memory_types?: MemoryType[]; + min_importance?: number; + channel_ids?: string[]; + max_age?: string; + mode?: SearchMode; + sort_by?: SearchSort; + max_memories?: number; +} + +export interface TopicItem { + id: string; + agent_id: string; + title: string; + content: string; + criteria: TopicCriteria; + pin_ids: string[]; + channel_ids: string[]; + status: TopicStatus; + max_words: number; + last_memory_at: string | null; + last_synced_at: string | null; + created_at: string; + updated_at: string; +} + +export interface TopicVersion { + id: string; + topic_id: string; + content: string; + memory_count: number; + created_at: string; +} + +export interface TopicListResponse { + topics: TopicItem[]; +} + +export interface TopicResponse { + topic: TopicItem; +} + +export interface TopicActionResponse { + success: boolean; + message: string; +} + +export interface TopicVersionsResponse { + versions: TopicVersion[]; +} + +export interface CreateTopicRequest { + title: string; + criteria?: TopicCriteria; + channel_ids?: string[]; + pin_ids?: string[]; + status?: TopicStatus; + max_words?: number; +} + +export interface UpdateTopicRequest { + title?: string; + criteria?: TopicCriteria; + channel_ids?: string[]; + pin_ids?: string[]; + status?: TopicStatus; + max_words?: number; + content?: string; +} + // -- Messaging / Bindings Types -- export interface PlatformStatus { @@ -1999,6 +2076,54 @@ export const api = { return response.json() as Promise; }, + // Topics API + listTopics: (agentId: string, status?: TopicStatus) => { + const search = new URLSearchParams({ agent_id: agentId }); + if (status) search.set("status", status); + return fetchJson(`/agents/topics?${search}`); + }, + getTopic: (agentId: string, topicId: string) => + fetchJson(`/agents/topics/${encodeURIComponent(topicId)}?agent_id=${encodeURIComponent(agentId)}`), + createTopic: async (agentId: string, request: CreateTopicRequest): Promise => { + const response = await fetch(`${API_BASE}/agents/topics`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ ...request, agent_id: agentId }), + }); + if (!response.ok) throw new Error(`API error: ${response.status}`); + return response.json() as Promise; + }, + updateTopic: async (agentId: string, topicId: string, request: UpdateTopicRequest): Promise => { + const response = await fetch(`${API_BASE}/agents/topics/${encodeURIComponent(topicId)}`, { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ ...request, agent_id: agentId }), + }); + if (!response.ok) throw new Error(`API error: ${response.status}`); + return response.json() as Promise; + }, + deleteTopic: async (agentId: string, topicId: string): Promise => { + const response = await fetch( + `${API_BASE}/agents/topics/${encodeURIComponent(topicId)}?agent_id=${encodeURIComponent(agentId)}`, + { method: "DELETE" }, + ); + if (!response.ok) throw new Error(`API error: ${response.status}`); + return response.json() as Promise; + }, + syncTopic: async (agentId: string, topicId: string): Promise => { + const response = await fetch(`${API_BASE}/agents/topics/${encodeURIComponent(topicId)}/sync`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ agent_id: agentId }), + }); + if (!response.ok) throw new Error(`API error: ${response.status}`); + return response.json() as Promise; + }, + topicVersions: (agentId: string, topicId: string, limit = 20) => + fetchJson( + `/agents/topics/${encodeURIComponent(topicId)}/versions?agent_id=${encodeURIComponent(agentId)}&limit=${limit}`, + ), + // Secrets API secretsStatus: () => fetchJson("/secrets/status"), listSecrets: () => fetchJson("/secrets"), diff --git a/interface/src/components/AgentTabs.tsx b/interface/src/components/AgentTabs.tsx index 1f0aaa8b4..523220362 100644 --- a/interface/src/components/AgentTabs.tsx +++ b/interface/src/components/AgentTabs.tsx @@ -9,6 +9,7 @@ const tabs = [ { label: "Ingest", to: "/agents/$agentId/ingest" as const, exact: false }, { label: "Workers", to: "/agents/$agentId/workers" as const, exact: false }, { label: "Tasks", to: "/agents/$agentId/tasks" as const, exact: false }, + { label: "Topics", to: "/agents/$agentId/topics" as const, exact: false }, { label: "Cortex", to: "/agents/$agentId/cortex" as const, exact: false }, { label: "Skills", to: "/agents/$agentId/skills" as const, exact: false }, { label: "Cron", to: "/agents/$agentId/cron" as const, exact: false }, diff --git a/interface/src/router.tsx b/interface/src/router.tsx index d65ce23fe..2d23fb579 100644 --- a/interface/src/router.tsx +++ b/interface/src/router.tsx @@ -21,6 +21,7 @@ import {AgentIngest} from "@/routes/AgentIngest"; import {AgentSkills} from "@/routes/AgentSkills"; import {AgentWorkers} from "@/routes/AgentWorkers"; import {AgentTasks} from "@/routes/AgentTasks"; +import {AgentTopics} from "@/routes/AgentTopics"; import {AgentChat} from "@/routes/AgentChat"; import {Settings} from "@/routes/Settings"; import {useLiveContext} from "@/hooks/useLiveContext"; @@ -219,6 +220,22 @@ const agentTasksRoute = createRoute({ }, }); +const agentTopicsRoute = createRoute({ + getParentRoute: () => rootRoute, + path: "/agents/$agentId/topics", + component: function AgentTopicsPage() { + const {agentId} = agentTopicsRoute.useParams(); + return ( +
+ +
+ +
+
+ ); + }, +}); + const agentCronRoute = createRoute({ getParentRoute: () => rootRoute, path: "/agents/$agentId/cron", @@ -323,6 +340,7 @@ const routeTree = rootRoute.addChildren([ agentIngestRoute, agentWorkersRoute, agentTasksRoute, + agentTopicsRoute, agentCortexRoute, agentSkillsRoute, agentCronRoute, diff --git a/interface/src/routes/AgentTopics.tsx b/interface/src/routes/AgentTopics.tsx new file mode 100644 index 000000000..7bddb702b --- /dev/null +++ b/interface/src/routes/AgentTopics.tsx @@ -0,0 +1,1028 @@ +import { useCallback, useState } from "react"; +import { useQuery, useMutation, useQueryClient } from "@tanstack/react-query"; +import { + api, + type TopicItem, + type TopicStatus, + type TopicVersion, + type CreateTopicRequest, + type UpdateTopicRequest, + type TopicCriteria, + type ChannelInfo, + type MemoryItem, + MEMORY_TYPES, + type MemoryType, +} from "@/api/client"; +import { Badge } from "@/ui/Badge"; +import { Button } from "@/ui/Button"; +import { + Dialog, + DialogContent, + DialogHeader, + DialogTitle, + DialogFooter, +} from "@/ui/Dialog"; +import { Markdown } from "@/components/Markdown"; +import { formatTimeAgo } from "@/lib/format"; + +const STATUS_COLORS: Record< + TopicStatus, + "green" | "amber" | "default" +> = { + active: "green", + paused: "amber", + archived: "default", +}; + +const MAX_AGE_OPTIONS = [ + { value: "", label: "All time" }, + { value: "7d", label: "7 days" }, + { value: "30d", label: "30 days" }, + { value: "90d", label: "90 days" }, +]; + +export function AgentTopics({ agentId }: { agentId: string }) { + const queryClient = useQueryClient(); + + const { data, isLoading } = useQuery({ + queryKey: ["topics", agentId], + queryFn: () => api.listTopics(agentId), + refetchInterval: 15_000, + }); + + const { data: channelsData } = useQuery({ + queryKey: ["channels"], + queryFn: api.channels, + }); + + const agentChannels = (channelsData?.channels ?? []).filter( + (c) => c.agent_id === agentId, + ); + + const topics = data?.topics ?? []; + + const [createOpen, setCreateOpen] = useState(false); + const [selectedTopicId, setSelectedTopicId] = useState(null); + const selectedTopic = selectedTopicId + ? (topics.find((t) => t.id === selectedTopicId) ?? null) + : null; + + const createMutation = useMutation({ + mutationFn: (request: CreateTopicRequest) => + api.createTopic(agentId, request), + onSuccess: () => { + queryClient.invalidateQueries({ queryKey: ["topics", agentId] }); + setCreateOpen(false); + }, + }); + + const updateMutation = useMutation({ + mutationFn: ({ + topicId, + ...request + }: UpdateTopicRequest & { topicId: string }) => + api.updateTopic(agentId, topicId, request), + onSuccess: () => { + queryClient.invalidateQueries({ queryKey: ["topics", agentId] }); + }, + }); + + const deleteMutation = useMutation({ + mutationFn: (topicId: string) => api.deleteTopic(agentId, topicId), + onSuccess: () => { + queryClient.invalidateQueries({ queryKey: ["topics", agentId] }); + setSelectedTopicId(null); + }, + }); + + const syncMutation = useMutation({ + mutationFn: (topicId: string) => api.syncTopic(agentId, topicId), + onSuccess: () => { + queryClient.invalidateQueries({ queryKey: ["topics", agentId] }); + }, + }); + + if (isLoading) { + return ( +
+ Loading topics... +
+ ); + } + + if (topics.length === 0 && !createOpen) { + return ( +
+

No topics yet

+

+ Topics are living context documents synthesized from memories. They + replace the global bulletin with targeted, per-channel context. +

+ + setCreateOpen(false)} + onCreate={(request) => createMutation.mutate(request)} + isPending={createMutation.isPending} + channels={agentChannels} + /> +
+ ); + } + + return ( +
+ {/* Toolbar */} +
+
+ + {topics.length} topic{topics.length !== 1 ? "s" : ""} + + {topics.filter((t) => t.status === "active").length > 0 && ( + + {topics.filter((t) => t.status === "active").length} active + + )} +
+ +
+ + {/* Card Grid */} +
+
+ {topics.map((topic) => ( + setSelectedTopicId(topic.id)} + onSync={() => syncMutation.mutate(topic.id)} + /> + ))} +
+
+ + {/* Create Dialog */} + setCreateOpen(false)} + onCreate={(request) => createMutation.mutate(request)} + isPending={createMutation.isPending} + channels={agentChannels} + /> + + {/* Detail Dialog */} + {selectedTopic && ( + setSelectedTopicId(null)} + onUpdate={(request) => + updateMutation.mutate({ + topicId: selectedTopic.id, + ...request, + }) + } + onDelete={() => deleteMutation.mutate(selectedTopic.id)} + onSync={() => syncMutation.mutate(selectedTopic.id)} + isSyncing={syncMutation.isPending} + /> + )} +
+ ); +} + +// -- Topic Card -- + +function TopicCard({ + topic, + onSelect, + onSync, +}: { + topic: TopicItem; + onSelect: () => void; + onSync: () => void; +}) { + const wordCount = topic.content + ? topic.content.split(/\s+/).filter(Boolean).length + : 0; + + return ( +
+ {/* Header */} +
+

+ {topic.title} +

+ + {topic.status} + +
+ + {/* Content preview */} + {topic.content && ( +

+ {topic.content.slice(0, 200)} +

+ )} + + {/* Stats row */} +
+ {topic.last_synced_at && ( + Synced {formatTimeAgo(topic.last_synced_at)} + )} + {!topic.last_synced_at && Never synced} + {topic.channel_ids.length > 0 && ( + + {topic.channel_ids.length} channel + {topic.channel_ids.length !== 1 ? "s" : ""} + + )} + {wordCount > 0 && {wordCount} words} +
+ + {/* Quick actions */} +
e.stopPropagation()}> + +
+
+ ); +} + +// -- Create Topic Dialog -- + +function CreateTopicDialog({ + open, + onClose, + onCreate, + isPending, + channels, +}: { + open: boolean; + onClose: () => void; + onCreate: (request: CreateTopicRequest) => void; + isPending: boolean; + channels: ChannelInfo[]; +}) { + const [title, setTitle] = useState(""); + const [query, setQuery] = useState(""); + const [selectedTypes, setSelectedTypes] = useState([]); + const [maxAge, setMaxAge] = useState(""); + const [maxWords, setMaxWords] = useState(1500); + const [selectedChannels, setSelectedChannels] = useState([]); + + const handleSubmit = useCallback(() => { + if (!title.trim()) return; + const criteria: TopicCriteria = {}; + if (query.trim()) criteria.query = query.trim(); + if (selectedTypes.length > 0) criteria.memory_types = selectedTypes; + if (maxAge) criteria.max_age = maxAge; + onCreate({ + title: title.trim(), + criteria: Object.keys(criteria).length > 0 ? criteria : undefined, + channel_ids: selectedChannels.length > 0 ? selectedChannels : undefined, + max_words: maxWords, + }); + setTitle(""); + setQuery(""); + setSelectedTypes([]); + setMaxAge(""); + setMaxWords(1500); + setSelectedChannels([]); + }, [title, query, selectedTypes, maxAge, maxWords, selectedChannels, onCreate]); + + return ( + !v && onClose()}> + + + Create Topic + +
+
+ + setTitle(e.target.value)} + onKeyDown={(e) => e.key === "Enter" && handleSubmit()} + autoFocus + /> +
+
+ + setQuery(e.target.value)} + /> +
+
+ +
+ {MEMORY_TYPES.map((type) => ( + + ))} +
+
+ {channels.length > 0 && ( +
+ +
+ {channels.map((channel) => ( + + ))} +
+
+ )} +
+
+ + +
+
+ + setMaxWords(Number(e.target.value))} + min={100} + max={10000} + /> +
+
+
+ + + + + + + ); +} + +// -- Topic Detail Dialog -- + +function TopicDetailDialog({ + agentId, + topic, + channels, + onClose, + onUpdate, + onDelete, + onSync, + isSyncing, +}: { + agentId: string; + topic: TopicItem; + channels: ChannelInfo[]; + onClose: () => void; + onUpdate: (request: UpdateTopicRequest) => void; + onDelete: () => void; + onSync: () => void; + isSyncing: boolean; +}) { + const [activeTab, setActiveTab] = useState<"content" | "config" | "versions">( + "content", + ); + + const wordCount = topic.content + ? topic.content.split(/\s+/).filter(Boolean).length + : 0; + + return ( + !v && onClose()}> + + +
+ {topic.title} + + {topic.status} + +
+ + + {/* Tabs */} +
+ {(["content", "config", "versions"] as const).map((tab) => ( + + ))} +
+ + {/* Tab content */} +
+ {activeTab === "content" && ( + + )} + {activeTab === "config" && ( + + )} + {activeTab === "versions" && ( + + )} +
+ + +
+ + +
+ + + + ); +} + +// -- Content Tab -- + +function ContentTab({ + topic, + wordCount, + onSync, + isSyncing, +}: { + topic: TopicItem; + wordCount: number; + onSync: () => void; + isSyncing: boolean; +}) { + return ( +
+ {/* Meta bar */} +
+
+ {wordCount} words + {topic.last_synced_at && ( + Last synced {formatTimeAgo(topic.last_synced_at)} + )} + {topic.channel_ids.length > 0 && ( + + {topic.channel_ids.length} channel + {topic.channel_ids.length !== 1 ? "s" : ""} + + )} +
+ +
+ + {/* Content */} + {topic.content ? ( +
+ {topic.content} +
+ ) : ( +
+ No content yet. Click "Sync Now" to generate. +
+ )} +
+ ); +} + +// -- Config Tab -- + +function ConfigTab({ + agentId, + topic, + channels, + onUpdate, +}: { + agentId: string; + topic: TopicItem; + channels: ChannelInfo[]; + onUpdate: (request: UpdateTopicRequest) => void; +}) { + const [title, setTitle] = useState(topic.title); + const [status, setStatus] = useState(topic.status); + const [query, setQuery] = useState(topic.criteria.query ?? ""); + const [selectedTypes, setSelectedTypes] = useState( + topic.criteria.memory_types ?? [], + ); + const [minImportance, setMinImportance] = useState( + topic.criteria.min_importance ?? 0, + ); + const [maxAge, setMaxAge] = useState(topic.criteria.max_age ?? ""); + const [maxMemories, setMaxMemories] = useState( + topic.criteria.max_memories ?? 30, + ); + const [maxWords, setMaxWords] = useState(topic.max_words); + const [selectedChannels, setSelectedChannels] = useState( + topic.channel_ids, + ); + const [pinIds, setPinIds] = useState(topic.pin_ids); + const [pinSearch, setPinSearch] = useState(""); + const [pinResults, setPinResults] = useState([]); + const [isSearching, setIsSearching] = useState(false); + + const handlePinSearch = useCallback(async () => { + if (!pinSearch.trim()) { + setPinResults([]); + return; + } + setIsSearching(true); + try { + const response = await api.searchMemories(agentId, pinSearch.trim(), { + limit: 10, + }); + setPinResults(response.results.map((r) => r.memory)); + } catch { + setPinResults([]); + } finally { + setIsSearching(false); + } + }, [agentId, pinSearch]); + + const hasChanges = + title !== topic.title || + status !== topic.status || + query !== (topic.criteria.query ?? "") || + JSON.stringify(selectedTypes) !== + JSON.stringify(topic.criteria.memory_types ?? []) || + minImportance !== (topic.criteria.min_importance ?? 0) || + maxAge !== (topic.criteria.max_age ?? "") || + maxMemories !== (topic.criteria.max_memories ?? 30) || + maxWords !== topic.max_words || + JSON.stringify(selectedChannels.slice().sort()) !== + JSON.stringify(topic.channel_ids.slice().sort()) || + JSON.stringify(pinIds.slice().sort()) !== + JSON.stringify(topic.pin_ids.slice().sort()); + + const handleSave = () => { + const criteria: TopicCriteria = { + max_memories: maxMemories, + }; + if (query.trim()) criteria.query = query.trim(); + if (selectedTypes.length > 0) criteria.memory_types = selectedTypes; + if (minImportance > 0) criteria.min_importance = minImportance; + if (maxAge) criteria.max_age = maxAge; + + onUpdate({ + title: title !== topic.title ? title : undefined, + status: status !== topic.status ? status : undefined, + criteria, + max_words: maxWords !== topic.max_words ? maxWords : undefined, + channel_ids: selectedChannels, + pin_ids: pinIds, + }); + }; + + return ( +
+ {/* Title */} +
+ + setTitle(e.target.value)} + /> +
+ + {/* Status */} +
+ +
+ {(["active", "paused", "archived"] as TopicStatus[]).map((s) => ( + + ))} +
+
+ + {/* Channel Assignments */} + {channels.length > 0 && ( +
+ +
+ {channels.map((channel) => ( + + ))} +
+
+ )} + + {/* Semantic Query */} +
+ + setQuery(e.target.value)} + /> +
+ + {/* Memory Types */} +
+ +
+ {MEMORY_TYPES.map((type) => ( + + ))} +
+
+ + {/* Min Importance */} +
+ + setMinImportance(Number(e.target.value))} + className="w-full accent-accent" + /> +
+ + {/* Max Age + Max Memories + Max Words */} +
+
+ + +
+
+ + setMaxMemories(Number(e.target.value))} + min={5} + max={100} + /> +
+
+ + setMaxWords(Number(e.target.value))} + min={100} + max={10000} + /> +
+
+ + {/* Pinned Memories */} +
+ +

+ Pinned memories are always included in the topic synthesis, + regardless of search criteria. +

+ + {/* Current pins */} + {pinIds.length > 0 && ( +
+ {pinIds.map((pinId) => ( +
+ + {pinId} + + +
+ ))} +
+ )} + + {/* Search to add pins */} +
+ setPinSearch(e.target.value)} + onKeyDown={(e) => e.key === "Enter" && handlePinSearch()} + /> + +
+ + {/* Search results */} + {pinResults.length > 0 && ( +
+ {pinResults + .filter((memory) => !pinIds.includes(memory.id)) + .map((memory) => ( + + ))} +
+ )} +
+ + {/* Save */} + {hasChanges && ( +
+ +
+ )} +
+ ); +} + +// -- Versions Tab -- + +function VersionsTab({ + agentId, + topicId, +}: { + agentId: string; + topicId: string; +}) { + const { data, isLoading } = useQuery({ + queryKey: ["topic-versions", agentId, topicId], + queryFn: () => api.topicVersions(agentId, topicId, 50), + }); + + const versions = data?.versions ?? []; + const [selectedVersion, setSelectedVersion] = useState( + null, + ); + + if (isLoading) { + return ( +
+ Loading versions... +
+ ); + } + + if (versions.length === 0) { + return ( +
+ No versions yet. Sync the topic to create the first version. +
+ ); + } + + return ( +
+ {/* Version list */} +
+ {versions.map((version, index) => ( + + ))} +
+ + {/* Selected version content */} + {selectedVersion && ( +
+
+ {formatTimeAgo(selectedVersion.created_at)} + {selectedVersion.memory_count} memories + + {selectedVersion.content.split(/\s+/).filter(Boolean).length}{" "} + words + +
+ + {selectedVersion.content} + +
+ )} +
+ ); +} diff --git a/migrations/20260302000001_topics.sql b/migrations/20260302000001_topics.sql new file mode 100644 index 000000000..399d9dd2f --- /dev/null +++ b/migrations/20260302000001_topics.sql @@ -0,0 +1,31 @@ +-- Topic synthesis: addressable, living context documents maintained by the cortex. + +CREATE TABLE IF NOT EXISTS topics ( + id TEXT PRIMARY KEY, + agent_id TEXT NOT NULL, + title TEXT NOT NULL, + content TEXT NOT NULL DEFAULT '', + criteria TEXT NOT NULL, + pin_ids TEXT NOT NULL DEFAULT '[]', + channel_ids TEXT NOT NULL DEFAULT '[]', + status TEXT NOT NULL DEFAULT 'active', + max_words INTEGER NOT NULL DEFAULT 1500, + last_memory_at TEXT, + last_synced_at TEXT, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +CREATE INDEX IF NOT EXISTS idx_topics_agent ON topics(agent_id); +CREATE INDEX IF NOT EXISTS idx_topics_status ON topics(status); + +CREATE TABLE IF NOT EXISTS topic_versions ( + id TEXT PRIMARY KEY, + topic_id TEXT NOT NULL, + content TEXT NOT NULL, + memory_count INTEGER NOT NULL, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY (topic_id) REFERENCES topics(id) ON DELETE CASCADE +); + +CREATE INDEX IF NOT EXISTS idx_topic_versions_topic ON topic_versions(topic_id); diff --git a/prompts/en/cortex_topic_synthesis.md.j2 b/prompts/en/cortex_topic_synthesis.md.j2 new file mode 100644 index 000000000..cf2ea4a6e --- /dev/null +++ b/prompts/en/cortex_topic_synthesis.md.j2 @@ -0,0 +1,21 @@ +You are synthesizing a focused context document about: {{ title }} + +You receive pre-gathered memory data matching this topic's criteria. Your job is to synthesize it into a concise, targeted briefing that will be injected into channels and workers that need context on this specific subject. + +## Output Format + +Write a focused document about the topic. Organize by what's most actionable or currently relevant. Use short paragraphs and clear structure. + +Do NOT: +- List raw memory IDs or metadata +- Reproduce the section headers from the input +- Repeat the same information in different phrasings +- Include information unrelated to the topic +- Exceed the word limit + +Do: +- Prioritize recent and high-importance information +- Connect related facts into coherent narratives +- Note any active contradictions or open questions +- Keep it scannable — short paragraphs, not walls of text +- Merge duplicates (the same memory may appear multiple times) diff --git a/src/agent/channel.rs b/src/agent/channel.rs index 7ced52b81..1897b9950 100644 --- a/src/agent/channel.rs +++ b/src/agent/channel.rs @@ -674,7 +674,25 @@ impl Channel { let prompt_engine = rc.prompts.load(); let identity_context = rc.identity.load().render(); - let memory_bulletin = rc.memory_bulletin.load(); + + // Use channel-specific topics if assigned, otherwise fall back to the global bulletin. + let channel_topics = self + .deps + .topic_store + .get_for_channel(&self.deps.agent_id, &self.id) + .await + .unwrap_or_default(); + let memory_bulletin = if channel_topics.is_empty() { + rc.memory_bulletin.load().to_string() + } else { + channel_topics + .iter() + .filter(|topic| !topic.content.is_empty()) + .map(|topic| format!("### {}\n\n{}", topic.title, topic.content)) + .collect::>() + .join("\n\n") + }; + let skills = rc.skills.load(); let skills_prompt = skills.render_channel_prompt(&prompt_engine)?; @@ -1002,7 +1020,44 @@ impl Channel { let prompt_engine = rc.prompts.load(); let identity_context = rc.identity.load().render(); - let memory_bulletin = rc.memory_bulletin.load(); + + // Use channel-specific topics if assigned, otherwise fall back to the global bulletin. + let channel_topics = self + .deps + .topic_store + .get_for_channel(&self.deps.agent_id, &self.id) + .await + .unwrap_or_default(); + let memory_bulletin = if channel_topics.is_empty() { + rc.memory_bulletin.load().to_string() + } else { + channel_topics + .iter() + .filter(|topic| !topic.content.is_empty()) + .map(|topic| format!("### {}\n\n{}", topic.title, topic.content)) + .collect::>() + .join("\n\n") + }; + + // List all active topics so the LLM can reference topic IDs when spawning workers. + let all_active_topics = self + .deps + .topic_store + .list_active(&self.deps.agent_id) + .await + .unwrap_or_default(); + let available_topics = if all_active_topics.is_empty() { + String::new() + } else { + let mut section = String::from( + "\n## Available Topics\nYou can pass these topic IDs to `spawn_worker` via the `topic_ids` parameter:\n", + ); + for topic in &all_active_topics { + section.push_str(&format!("- `{}`: {}\n", topic.id, topic.title)); + } + section + }; + let skills = rc.skills.load(); let skills_prompt = skills.render_channel_prompt(&prompt_engine)?; @@ -1020,7 +1075,9 @@ impl Channel { let current_time_line = temporal_context.current_time_line(); let status_text = { let status = self.state.status_block.read().await; - status.render_with_time_context(Some(¤t_time_line)) + let mut rendered = status.render_with_time_context(Some(¤t_time_line)); + rendered.push_str(&available_topics); + rendered }; let available_channels = self.build_available_channels().await; diff --git a/src/agent/channel_dispatch.rs b/src/agent/channel_dispatch.rs index 42b24aed7..36fdcecd2 100644 --- a/src/agent/channel_dispatch.rs +++ b/src/agent/channel_dispatch.rs @@ -240,6 +240,7 @@ pub async fn spawn_worker_from_state( task: impl Into, interactive: bool, suggested_skills: &[&str], + topic_ids: &[String], ) -> std::result::Result { check_worker_limit(state).await?; ensure_dispatch_readiness(state, "worker"); @@ -248,9 +249,26 @@ pub async fn spawn_worker_from_state( let rc = &state.deps.runtime_config; let prompt_engine = rc.prompts.load(); let temporal_context = TemporalContext::from_runtime(rc.as_ref()); - let worker_task = + let mut worker_task = build_worker_task_with_temporal_context(&task, &temporal_context, &prompt_engine) .map_err(|error| AgentError::Other(anyhow::anyhow!("{error}")))?; + + // Inject topic context if topic_ids were provided. + if !topic_ids.is_empty() { + let mut topic_context = String::from("## Topic Context\n\n"); + for topic_id in topic_ids { + if let Ok(Some(topic)) = state.deps.topic_store.get(topic_id).await { + if !topic.content.is_empty() { + topic_context + .push_str(&format!("### {}\n\n{}\n\n", topic.title, topic.content)); + } + } + } + if topic_context.len() > "## Topic Context\n\n".len() { + worker_task = format!("{topic_context}---\n\n{worker_task}"); + } + } + let sandbox_enabled = state.deps.sandbox.mode_enabled(); let sandbox_containment_active = state.deps.sandbox.containment_active(); let sandbox_read_allowlist = state.deps.sandbox.prompt_read_allowlist(); diff --git a/src/agent/cortex.rs b/src/agent/cortex.rs index c645fccf0..c41f12d6d 100644 --- a/src/agent/cortex.rs +++ b/src/agent/cortex.rs @@ -339,6 +339,8 @@ impl Cortex { /// Process a process event and extract signals. pub async fn observe(&self, event: ProcessEvent) { + let is_memory_saved = matches!(&event, ProcessEvent::MemorySaved { .. }); + let signal = match &event { ProcessEvent::MemorySaved { memory_id, .. } => Some(Signal::MemorySaved { memory_type: "unknown".into(), @@ -370,6 +372,19 @@ impl Cortex { tracing::debug!("cortex received signal, buffer size: {}", buffer.len()); } + + // When a memory is saved, mark all active topics as stale so the + // topic sync loop re-synthesizes them on its next pass. + if is_memory_saved { + if let Err(error) = self + .deps + .topic_store + .mark_all_stale(&self.deps.agent_id) + .await + { + tracing::warn!(%error, "failed to mark topics stale after memory save"); + } + } } /// Run periodic consolidation (future: health monitoring, memory maintenance). @@ -1201,6 +1216,26 @@ async fn pickup_one_ready_task(deps: &AgentDeps, logger: &CortexLogger) -> anyho } } + // Inject topic content from task metadata if topic_ids are present. + if let Some(topic_ids) = task.metadata.get("topic_ids").and_then(|v| v.as_array()) { + let topic_ids: Vec<&str> = topic_ids.iter().filter_map(|v| v.as_str()).collect(); + if !topic_ids.is_empty() { + let mut topic_content = String::new(); + for topic_id in &topic_ids { + if let Ok(Some(topic)) = deps.topic_store.get(topic_id).await { + if !topic.content.is_empty() { + topic_content + .push_str(&format!("\n## {}\n{}\n", topic.title, topic.content)); + } + } + } + if !topic_content.is_empty() { + task_prompt.push_str("\n\n---\nRelevant Context:\n"); + task_prompt.push_str(&topic_content); + } + } + } + let screenshot_dir = deps .runtime_config .workspace_dir @@ -1679,6 +1714,231 @@ async fn fetch_memories_for_association( Ok(rows.iter().map(|row| row.get("id")).collect()) } +// --------------------------------------------------------------------------- +// Topic Sync Loop +// --------------------------------------------------------------------------- + +/// Spawn the topic sync loop. Runs alongside bulletin and association loops. +pub fn spawn_topic_sync_loop(deps: AgentDeps, logger: CortexLogger) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + if let Err(error) = run_topic_sync_loop(&deps, &logger).await { + tracing::error!(%error, "cortex topic sync loop exited with error"); + } + }) +} + +async fn run_topic_sync_loop(deps: &AgentDeps, logger: &CortexLogger) -> anyhow::Result<()> { + tracing::info!("cortex topic sync loop started"); + + // Initial sync on startup — sync all active topics. + if let Err(error) = sync_all_topics(deps, logger).await { + tracing::warn!(%error, "initial topic sync pass failed"); + } + + loop { + let interval = deps.runtime_config.cortex.load().topic_sync_interval_secs; + tokio::time::sleep(Duration::from_secs(interval.max(30))).await; + + if let Err(error) = sync_all_topics(deps, logger).await { + tracing::warn!(%error, "topic sync pass failed"); + } + } +} + +/// Run a sync pass over all active topics, only synthesizing stale ones. +async fn sync_all_topics(deps: &AgentDeps, logger: &CortexLogger) -> anyhow::Result<()> { + let topics = deps.topic_store.list_active(&deps.agent_id).await?; + if topics.is_empty() { + return Ok(()); + } + + let mut synced_count = 0usize; + for topic in &topics { + match sync_one_topic(deps, logger, topic).await { + Ok(true) => synced_count += 1, + Ok(false) => {} // Not stale, skipped. + Err(error) => { + tracing::warn!( + topic_id = %topic.id, + topic_title = %topic.title, + %error, + "topic sync failed" + ); + logger.log( + "topic_sync_failed", + &format!("Topic sync failed for '{}': {error}", topic.title), + Some(serde_json::json!({ + "topic_id": topic.id, + "title": topic.title, + })), + ); + } + } + } + + if synced_count > 0 { + tracing::info!( + synced = synced_count, + total = topics.len(), + "topic sync pass complete" + ); + } + + Ok(()) +} + +/// Sync a single topic: check staleness, gather memories, synthesize via LLM, +/// update content + version. Returns `true` if the topic was synced. +async fn sync_one_topic( + deps: &AgentDeps, + logger: &CortexLogger, + topic: &crate::topics::Topic, +) -> anyhow::Result { + use anyhow::Context as _; + + let search_config = topic.criteria.to_search_config(); + let query = topic.criteria.query.as_deref().unwrap_or(""); + + // Staleness check: query newest memory matching criteria and compare to last_memory_at. + let newest_results = deps + .memory_search + .search( + query, + &SearchConfig { + max_results: 1, + sort_by: SearchSort::Recent, + ..search_config.clone() + }, + ) + .await + .unwrap_or_default(); + + let newest_memory_at = newest_results + .first() + .map(|result| result.memory.created_at.to_rfc3339()); + + let is_stale = topic.last_synced_at.is_none() + || match (&newest_memory_at, &topic.last_memory_at) { + (Some(newest), Some(last)) => newest.as_str() > last.as_str(), + (Some(_), None) => true, + _ => false, + }; + + if !is_stale { + return Ok(false); + } + + // Gather memories matching criteria. + let mut results = deps + .memory_search + .search(query, &search_config) + .await + .unwrap_or_default(); + + // Merge pinned memories (prepend, dedup). + if !topic.pin_ids.is_empty() { + let store = deps.memory_search.store(); + for pin_id in &topic.pin_ids { + if let Ok(Some(memory)) = store.load(pin_id).await { + // Only add if not already in results. + if !results.iter().any(|r| r.memory.id == memory.id) { + results.insert( + 0, + crate::memory::types::MemorySearchResult { + memory, + score: 1.0, + rank: 0, + }, + ); + } + } + } + } + + if results.is_empty() { + return Ok(false); + } + + let memory_count = results.len(); + + // Format as raw sections. + let mut raw_sections = String::new(); + for result in &results { + raw_sections.push_str(&format!( + "- [{}] (importance: {:.1}) {}\n", + result.memory.memory_type, + result.memory.importance, + result + .memory + .content + .lines() + .next() + .unwrap_or(&result.memory.content), + )); + } + + // LLM synthesis. + let prompt_engine = deps.runtime_config.prompts.load(); + let system_prompt = prompt_engine + .render_topic_synthesis_prompt(&topic.title) + .context("failed to render topic synthesis prompt")?; + + let synthesis_prompt = prompt_engine + .render_system_cortex_synthesis(topic.max_words, &raw_sections) + .context("failed to render cortex synthesis user prompt")?; + + let routing = deps.runtime_config.routing.load(); + let model_name = routing.resolve(ProcessType::Branch, None).to_string(); + let model = SpacebotModel::make(&deps.llm_manager, &model_name) + .with_context(&*deps.agent_id, "cortex") + .with_routing((**routing).clone()); + + let agent = rig::agent::AgentBuilder::new(model) + .preamble(&system_prompt) + .build(); + + let content = agent + .prompt(&synthesis_prompt) + .await + .map_err(|error| anyhow::anyhow!("topic synthesis LLM call failed: {error}"))?; + + // Update topic content and timestamps. + let now = chrono::Utc::now().to_rfc3339(); + let mut updated_topic = topic.clone(); + updated_topic.content = content.clone(); + updated_topic.last_synced_at = Some(now.clone()); + updated_topic.last_memory_at = newest_memory_at; + deps.topic_store.update(&updated_topic).await?; + + // Save version snapshot. + let version = crate::topics::TopicVersion { + id: uuid::Uuid::new_v4().to_string(), + topic_id: topic.id.clone(), + content, + memory_count: memory_count as i64, + created_at: now, + }; + deps.topic_store.save_version(&version).await?; + + let word_count = updated_topic.content.split_whitespace().count(); + logger.log( + "topic_synced", + &format!( + "Topic '{}' synced: {word_count} words from {memory_count} memories", + topic.title + ), + Some(serde_json::json!({ + "topic_id": topic.id, + "title": topic.title, + "word_count": word_count, + "memory_count": memory_count, + "model": model_name, + })), + ); + + Ok(true) +} + #[cfg(test)] mod tests { use super::{ diff --git a/src/agent/cortex_chat.rs b/src/agent/cortex_chat.rs index ed5ec0033..41bb24c50 100644 --- a/src/agent/cortex_chat.rs +++ b/src/agent/cortex_chat.rs @@ -328,7 +328,25 @@ impl CortexChatSession { let prompt_engine = runtime_config.prompts.load(); let identity_context = runtime_config.identity.load().render(); - let memory_bulletin = runtime_config.memory_bulletin.load(); + + // Compose all active topics for the cortex chat context, falling + // back to the global bulletin if no topics exist. + let all_topics = self + .deps + .topic_store + .list_active(&self.deps.agent_id) + .await + .unwrap_or_default(); + let context_content = if all_topics.is_empty() { + runtime_config.memory_bulletin.load().to_string() + } else { + all_topics + .iter() + .filter(|topic| !topic.content.is_empty()) + .map(|topic| format!("### {}\n\n{}", topic.title, topic.content)) + .collect::>() + .join("\n\n") + }; let browser_enabled = runtime_config.browser_config.load().enabled; let web_search_enabled = runtime_config.brave_search_key.load().is_some(); @@ -350,7 +368,7 @@ impl CortexChatSession { prompt_engine.render_cortex_chat_prompt( empty_to_none(identity_context), - empty_to_none(memory_bulletin.to_string()), + empty_to_none(context_content), channel_transcript, worker_capabilities, ) diff --git a/src/api.rs b/src/api.rs index 4d83da0b9..9ef2008d6 100644 --- a/src/api.rs +++ b/src/api.rs @@ -25,6 +25,7 @@ mod state; mod system; mod tasks; mod tools; +mod topics; mod webchat; mod workers; diff --git a/src/api/agents.rs b/src/api/agents.rs index 81999c552..121b72db0 100644 --- a/src/api/agents.rs +++ b/src/api/agents.rs @@ -416,6 +416,9 @@ pub(super) async fn trigger_warmup( messaging_manager: None, sandbox, task_store, + topic_store: std::sync::Arc::new(crate::topics::TopicStore::new( + sqlite_pool.clone(), + )), links: Arc::new(arc_swap::ArcSwap::from_pointee(Vec::new())), agent_names: Arc::new(std::collections::HashMap::new()), task_store_registry, @@ -628,6 +631,7 @@ pub(super) async fn create_agent( embedding_model, )); let task_store = std::sync::Arc::new(crate::tasks::TaskStore::new(db.sqlite.clone())); + let topic_store = std::sync::Arc::new(crate::topics::TopicStore::new(db.sqlite.clone())); let (event_tx, _) = tokio::sync::broadcast::channel(256); let arc_agent_id: crate::AgentId = std::sync::Arc::from(agent_id.as_str()); @@ -708,6 +712,7 @@ pub(super) async fn create_agent( llm_manager, mcp_manager: mcp_manager.clone(), task_store: task_store.clone(), + topic_store: topic_store.clone(), cron_tool: None, runtime_config: runtime_config.clone(), event_tx: event_tx.clone(), @@ -834,6 +839,12 @@ pub(super) async fn create_agent( task_stores.insert(agent_id.clone(), task_store.clone()); state.task_stores.store(std::sync::Arc::new(task_stores)); + let mut topic_stores_map = (**state.topic_stores.load()).clone(); + topic_stores_map.insert(agent_id.clone(), topic_store); + state + .topic_stores + .store(std::sync::Arc::new(topic_stores_map)); + let mut registry = (**state.task_store_registry.load()).clone(); registry.insert(agent_id.clone(), task_store); state diff --git a/src/api/server.rs b/src/api/server.rs index 24d2fc9e9..a84c9a68a 100644 --- a/src/api/server.rs +++ b/src/api/server.rs @@ -3,7 +3,7 @@ use super::state::ApiState; use super::{ agents, bindings, channels, config, cortex, cron, ingest, links, mcp, memories, messaging, - models, providers, secrets, settings, skills, system, tasks, tools, webchat, workers, + models, providers, secrets, settings, skills, system, tasks, tools, topics, webchat, workers, }; use axum::Json; @@ -130,6 +130,18 @@ pub async fn start_http_server( ) .route("/agents/tasks/{number}/approve", post(tasks::approve_task)) .route("/agents/tasks/{number}/execute", post(tasks::execute_task)) + .route( + "/agents/topics", + get(topics::list_topics).post(topics::create_topic), + ) + .route( + "/agents/topics/{id}", + get(topics::get_topic) + .put(topics::update_topic) + .delete(topics::delete_topic), + ) + .route("/agents/topics/{id}/sync", post(topics::sync_topic)) + .route("/agents/topics/{id}/versions", get(topics::topic_versions)) .route("/channels/cancel", post(channels::cancel_process)) .route( "/agents/ingest/files", diff --git a/src/api/state.rs b/src/api/state.rs index 1a00a5d71..8cebb1cfd 100644 --- a/src/api/state.rs +++ b/src/api/state.rs @@ -68,6 +68,8 @@ pub struct ApiState { pub cron_schedulers: arc_swap::ArcSwap>>, /// Per-agent task stores for task CRUD operations. pub task_stores: arc_swap::ArcSwap>>, + /// Per-agent topic stores for topic CRUD operations. + pub topic_stores: arc_swap::ArcSwap>>, /// Per-agent RuntimeConfig for reading live hot-reloaded configuration. pub runtime_configs: ArcSwap>>, /// Per-agent MCP managers for status and reconnect APIs. @@ -248,6 +250,7 @@ impl ApiState { cron_stores: arc_swap::ArcSwap::from_pointee(HashMap::new()), cron_schedulers: arc_swap::ArcSwap::from_pointee(HashMap::new()), task_stores: arc_swap::ArcSwap::from_pointee(HashMap::new()), + topic_stores: arc_swap::ArcSwap::from_pointee(HashMap::new()), runtime_configs: ArcSwap::from_pointee(HashMap::new()), mcp_managers: ArcSwap::from_pointee(HashMap::new()), sandboxes: ArcSwap::from_pointee(HashMap::new()), @@ -537,6 +540,11 @@ impl ApiState { self.task_stores.store(Arc::new(stores)); } + /// Set the topic stores for all agents. + pub fn set_topic_stores(&self, stores: HashMap>) { + self.topic_stores.store(Arc::new(stores)); + } + /// Set the runtime configs for all agents. pub fn set_runtime_configs(&self, configs: HashMap>) { self.runtime_configs.store(Arc::new(configs)); diff --git a/src/api/topics.rs b/src/api/topics.rs new file mode 100644 index 000000000..29ddeb6b9 --- /dev/null +++ b/src/api/topics.rs @@ -0,0 +1,296 @@ +//! Topic API handlers. + +use super::state::ApiState; + +use axum::Json; +use axum::extract::{Path, Query, State}; +use axum::http::StatusCode; +use serde::{Deserialize, Serialize}; +use std::sync::Arc; + +#[derive(Deserialize)] +pub(super) struct TopicListQuery { + agent_id: String, + #[serde(default)] + status: Option, +} + +#[derive(Deserialize)] +pub(super) struct TopicGetQuery { + agent_id: String, +} + +#[derive(Deserialize)] +pub(super) struct CreateTopicRequest { + agent_id: String, + title: String, + #[serde(default)] + criteria: Option, + #[serde(default)] + channel_ids: Vec, + #[serde(default)] + pin_ids: Vec, + #[serde(default)] + status: Option, + #[serde(default)] + max_words: Option, +} + +#[derive(Deserialize)] +pub(super) struct UpdateTopicRequest { + agent_id: String, + #[serde(default)] + title: Option, + #[serde(default)] + criteria: Option, + #[serde(default)] + channel_ids: Option>, + #[serde(default)] + pin_ids: Option>, + #[serde(default)] + status: Option, + #[serde(default)] + max_words: Option, + #[serde(default)] + content: Option, +} + +#[derive(Deserialize)] +pub(super) struct DeleteTopicQuery { + agent_id: String, +} + +#[derive(Deserialize)] +pub(super) struct SyncTopicRequest { + agent_id: String, +} + +#[derive(Deserialize)] +pub(super) struct VersionsQuery { + agent_id: String, + #[serde(default = "default_version_limit")] + limit: i64, +} + +fn default_version_limit() -> i64 { + 20 +} + +#[derive(Serialize)] +pub(super) struct TopicListResponse { + topics: Vec, +} + +#[derive(Serialize)] +pub(super) struct TopicResponse { + topic: crate::topics::Topic, +} + +#[derive(Serialize)] +pub(super) struct TopicActionResponse { + success: bool, + message: String, +} + +#[derive(Serialize)] +pub(super) struct TopicVersionsResponse { + versions: Vec, +} + +pub(super) async fn list_topics( + State(state): State>, + Query(query): Query, +) -> Result, StatusCode> { + let stores = state.topic_stores.load(); + let store = stores.get(&query.agent_id).ok_or(StatusCode::NOT_FOUND)?; + + let topics = if query.status.as_deref() == Some("active") { + store.list_active(&query.agent_id).await.map_err(|error| { + tracing::warn!(%error, agent_id = %query.agent_id, "failed to list active topics"); + StatusCode::INTERNAL_SERVER_ERROR + })? + } else { + store.list(&query.agent_id).await.map_err(|error| { + tracing::warn!(%error, agent_id = %query.agent_id, "failed to list topics"); + StatusCode::INTERNAL_SERVER_ERROR + })? + }; + + Ok(Json(TopicListResponse { topics })) +} + +pub(super) async fn get_topic( + State(state): State>, + Path(id): Path, + Query(query): Query, +) -> Result, StatusCode> { + let stores = state.topic_stores.load(); + let store = stores.get(&query.agent_id).ok_or(StatusCode::NOT_FOUND)?; + + let topic = store + .get(&id) + .await + .map_err(|error| { + tracing::warn!(%error, agent_id = %query.agent_id, topic_id = %id, "failed to get topic"); + StatusCode::INTERNAL_SERVER_ERROR + })? + .ok_or(StatusCode::NOT_FOUND)?; + + Ok(Json(TopicResponse { topic })) +} + +pub(super) async fn create_topic( + State(state): State>, + Json(request): Json, +) -> Result, StatusCode> { + let stores = state.topic_stores.load(); + let store = stores.get(&request.agent_id).ok_or(StatusCode::NOT_FOUND)?; + + let status = match request.status.as_deref() { + None | Some("active") => crate::topics::TopicStatus::Active, + Some("paused") => crate::topics::TopicStatus::Paused, + Some("archived") => crate::topics::TopicStatus::Archived, + Some(_) => return Err(StatusCode::BAD_REQUEST), + }; + + let topic = crate::topics::Topic { + id: uuid::Uuid::new_v4().to_string(), + agent_id: request.agent_id.clone(), + title: request.title, + content: String::new(), + criteria: request.criteria.unwrap_or_default(), + pin_ids: request.pin_ids, + channel_ids: request.channel_ids, + status, + max_words: request.max_words.unwrap_or(1500), + last_memory_at: None, + last_synced_at: None, + created_at: chrono::Utc::now().to_rfc3339(), + updated_at: chrono::Utc::now().to_rfc3339(), + }; + + store.create(&topic).await.map_err(|error| { + tracing::warn!(%error, agent_id = %request.agent_id, "failed to create topic"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + + Ok(Json(TopicResponse { topic })) +} + +pub(super) async fn update_topic( + State(state): State>, + Path(id): Path, + Json(request): Json, +) -> Result, StatusCode> { + let stores = state.topic_stores.load(); + let store = stores.get(&request.agent_id).ok_or(StatusCode::NOT_FOUND)?; + + let mut topic = store + .get(&id) + .await + .map_err(|error| { + tracing::warn!(%error, agent_id = %request.agent_id, topic_id = %id, "failed to get topic for update"); + StatusCode::INTERNAL_SERVER_ERROR + })? + .ok_or(StatusCode::NOT_FOUND)?; + + if let Some(title) = request.title { + topic.title = title; + } + if let Some(criteria) = request.criteria { + topic.criteria = criteria; + } + if let Some(channel_ids) = request.channel_ids { + topic.channel_ids = channel_ids; + } + if let Some(pin_ids) = request.pin_ids { + topic.pin_ids = pin_ids; + } + if let Some(status_str) = request.status { + topic.status = + crate::topics::TopicStatus::parse(&status_str).ok_or(StatusCode::BAD_REQUEST)?; + } + if let Some(max_words) = request.max_words { + topic.max_words = max_words; + } + if let Some(content) = request.content { + topic.content = content; + } + + store.update(&topic).await.map_err(|error| { + tracing::warn!(%error, agent_id = %request.agent_id, topic_id = %id, "failed to update topic"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + + Ok(Json(TopicResponse { topic })) +} + +pub(super) async fn delete_topic( + State(state): State>, + Path(id): Path, + Query(query): Query, +) -> Result, StatusCode> { + let stores = state.topic_stores.load(); + let store = stores.get(&query.agent_id).ok_or(StatusCode::NOT_FOUND)?; + + let deleted = store.delete(&id).await.map_err(|error| { + tracing::warn!(%error, agent_id = %query.agent_id, topic_id = %id, "failed to delete topic"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + + if !deleted { + return Err(StatusCode::NOT_FOUND); + } + + Ok(Json(TopicActionResponse { + success: true, + message: format!("Topic {id} deleted"), + })) +} + +pub(super) async fn sync_topic( + State(state): State>, + Path(id): Path, + Json(request): Json, +) -> Result, StatusCode> { + let stores = state.topic_stores.load(); + let store = stores.get(&request.agent_id).ok_or(StatusCode::NOT_FOUND)?; + + let topic = store + .get(&id) + .await + .map_err(|error| { + tracing::warn!(%error, "failed to get topic for sync"); + StatusCode::INTERNAL_SERVER_ERROR + })? + .ok_or(StatusCode::NOT_FOUND)?; + + // Clear last_synced_at to force a re-sync on the next topic sync pass. + let mut updated = topic; + updated.last_synced_at = None; + store.update(&updated).await.map_err(|error| { + tracing::warn!(%error, "failed to clear topic sync timestamp"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + + Ok(Json(TopicResponse { topic: updated })) +} + +pub(super) async fn topic_versions( + State(state): State>, + Path(id): Path, + Query(query): Query, +) -> Result, StatusCode> { + let stores = state.topic_stores.load(); + let store = stores.get(&query.agent_id).ok_or(StatusCode::NOT_FOUND)?; + + let versions = store + .get_versions(&id, query.limit) + .await + .map_err(|error| { + tracing::warn!(%error, topic_id = %id, "failed to get topic versions"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + + Ok(Json(TopicVersionsResponse { versions })) +} diff --git a/src/config.rs b/src/config.rs index c3a9ec8ac..5ddabc26a 100644 --- a/src/config.rs +++ b/src/config.rs @@ -985,6 +985,8 @@ pub struct CortexConfig { pub association_updates_threshold: f32, /// Max associations to create per pass (rate limit). pub association_max_per_pass: usize, + /// Interval in seconds between topic sync passes. + pub topic_sync_interval_secs: u64, } impl Default for CortexConfig { @@ -1001,6 +1003,7 @@ impl Default for CortexConfig { association_similarity_threshold: 0.85, association_updates_threshold: 0.95, association_max_per_pass: 100, + topic_sync_interval_secs: 600, } } } @@ -2972,6 +2975,7 @@ struct TomlCortexConfig { association_similarity_threshold: Option, association_updates_threshold: Option, association_max_per_pass: Option, + topic_sync_interval_secs: Option, } #[derive(Deserialize)] @@ -4807,6 +4811,9 @@ impl Config { association_max_per_pass: c .association_max_per_pass .unwrap_or(base_defaults.cortex.association_max_per_pass), + topic_sync_interval_secs: c + .topic_sync_interval_secs + .unwrap_or(base_defaults.cortex.topic_sync_interval_secs), }) .unwrap_or(base_defaults.cortex), warmup: toml @@ -5013,6 +5020,9 @@ impl Config { association_max_per_pass: c .association_max_per_pass .unwrap_or(defaults.cortex.association_max_per_pass), + topic_sync_interval_secs: c + .topic_sync_interval_secs + .unwrap_or(defaults.cortex.topic_sync_interval_secs), }), warmup: a.warmup.map(|w| WarmupConfig { enabled: w.enabled.unwrap_or(defaults.warmup.enabled), diff --git a/src/lib.rs b/src/lib.rs index 3557849b0..bc9144ac4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -27,6 +27,7 @@ pub mod tasks; #[cfg(feature = "metrics")] pub mod telemetry; pub mod tools; +pub mod topics; pub mod update; pub use error::{Error, Result}; @@ -218,6 +219,7 @@ pub struct AgentDeps { pub llm_manager: Arc, pub mcp_manager: Arc, pub task_store: Arc, + pub topic_store: Arc, pub cron_tool: Option, pub runtime_config: Arc, pub event_tx: tokio::sync::broadcast::Sender, diff --git a/src/main.rs b/src/main.rs index ae4dceae0..f0ffb72f3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2180,6 +2180,7 @@ async fn initialize_agents( llm_manager: llm_manager.clone(), mcp_manager, task_store: task_store.clone(), + topic_store: Arc::new(spacebot::topics::TopicStore::new(db.sqlite.clone())), cron_tool: None, runtime_config, event_tx, @@ -2246,6 +2247,7 @@ async fn initialize_agents( let mut memory_searches = std::collections::HashMap::new(); let mut mcp_managers = std::collections::HashMap::new(); let mut task_stores = std::collections::HashMap::new(); + let mut topic_stores = std::collections::HashMap::new(); let mut agent_workspaces = std::collections::HashMap::new(); let mut runtime_configs = std::collections::HashMap::new(); let mut sandboxes = std::collections::HashMap::new(); @@ -2256,6 +2258,7 @@ async fn initialize_agents( memory_searches.insert(agent_id.to_string(), agent.deps.memory_search.clone()); mcp_managers.insert(agent_id.to_string(), agent.deps.mcp_manager.clone()); task_stores.insert(agent_id.to_string(), agent.deps.task_store.clone()); + topic_stores.insert(agent_id.to_string(), agent.deps.topic_store.clone()); agent_workspaces.insert(agent_id.to_string(), agent.config.workspace.clone()); runtime_configs.insert(agent_id.to_string(), agent.deps.runtime_config.clone()); sandboxes.insert(agent_id.to_string(), agent.deps.sandbox.clone()); @@ -2275,6 +2278,7 @@ async fn initialize_agents( api_state.set_memory_searches(memory_searches); api_state.set_mcp_managers(mcp_managers); api_state.set_task_stores(task_stores); + api_state.set_topic_stores(topic_stores); api_state.set_runtime_configs(runtime_configs); api_state.set_agent_workspaces(agent_workspaces); api_state.set_sandboxes(sandboxes); @@ -2770,6 +2774,13 @@ async fn initialize_agents( ); cortex_handles.push(ready_task_handle); tracing::info!(agent_id = %agent_id, "cortex ready-task loop started"); + + let topic_sync_handle = spacebot::agent::cortex::spawn_topic_sync_loop( + agent.deps.clone(), + spacebot::agent::cortex::CortexLogger::new(agent.db.sqlite.clone()), + ); + cortex_handles.push(topic_sync_handle); + tracing::info!(agent_id = %agent_id, "cortex topic sync loop started"); } // Create cortex chat sessions for each agent diff --git a/src/memory/search.rs b/src/memory/search.rs index ae6e815f5..828ecf963 100644 --- a/src/memory/search.rs +++ b/src/memory/search.rs @@ -8,7 +8,8 @@ use std::collections::HashMap; use std::sync::Arc; /// Which search strategy to use. -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] pub enum SearchMode { /// Full hybrid: vector + FTS + graph + RRF. Requires a query string. #[default] @@ -22,7 +23,8 @@ pub enum SearchMode { } /// Sort order for non-hybrid search modes. -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] pub enum SearchSort { /// Most recent first (created_at DESC). #[default] diff --git a/src/prompts/engine.rs b/src/prompts/engine.rs index d24c677bf..aeed8bfb2 100644 --- a/src/prompts/engine.rs +++ b/src/prompts/engine.rs @@ -396,6 +396,16 @@ impl PromptEngine { ) } + /// Convenience method for rendering topic synthesis system prompt. + pub fn render_topic_synthesis_prompt(&self, title: &str) -> Result { + self.render( + "cortex_topic_synthesis", + context! { + title => title, + }, + ) + } + /// Convenience method for rendering ingestion chunk prompt. pub fn render_system_ingestion_chunk( &self, diff --git a/src/tools/spawn_worker.rs b/src/tools/spawn_worker.rs index 8edc4ccc5..b07dda91b 100644 --- a/src/tools/spawn_worker.rs +++ b/src/tools/spawn_worker.rs @@ -49,6 +49,10 @@ pub struct SpawnWorkerArgs { /// The OpenCode agent will operate in this directory. #[serde(default)] pub directory: Option, + /// Topic IDs to inject as context. The synthesized content of each topic + /// is prepended to the worker's task prompt. + #[serde(default)] + pub topic_ids: Vec, } /// Output from spawn worker tool. @@ -110,6 +114,11 @@ impl Tool for SpawnWorkerTool { "type": "array", "items": { "type": "string" }, "description": "Skill names from that are likely relevant to this task. The worker sees all skills and decides what to read, but suggested skills are flagged as recommended." + }, + "topic_ids": { + "type": "array", + "items": { "type": "string" }, + "description": "Topic IDs to inject as context into the worker. The synthesized content of each topic is prepended to the task. See the Available Topics section for IDs." } }); @@ -165,6 +174,7 @@ impl Tool for SpawnWorkerTool { .iter() .map(String::as_str) .collect::>(), + &args.topic_ids, ) .await .map_err(|e| SpawnWorkerError(format!("{e}")))? diff --git a/src/topics.rs b/src/topics.rs new file mode 100644 index 000000000..2dc75f3fa --- /dev/null +++ b/src/topics.rs @@ -0,0 +1,5 @@ +//! Topic synthesis: addressable, living context documents maintained by the cortex. + +pub mod store; + +pub use store::{Topic, TopicCriteria, TopicStatus, TopicStore, TopicVersion}; diff --git a/src/topics/store.rs b/src/topics/store.rs new file mode 100644 index 000000000..417eec3f7 --- /dev/null +++ b/src/topics/store.rs @@ -0,0 +1,605 @@ +//! Topic CRUD storage (SQLite). + +use crate::error::Result; +use crate::memory::{MemoryType, SearchConfig, SearchMode, SearchSort}; + +use anyhow::Context as _; +use serde::{Deserialize, Serialize}; +use sqlx::{Row as _, SqlitePool}; + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TopicStatus { + Active, + Paused, + Archived, +} + +impl TopicStatus { + pub fn as_str(self) -> &'static str { + match self { + TopicStatus::Active => "active", + TopicStatus::Paused => "paused", + TopicStatus::Archived => "archived", + } + } + + pub fn parse(value: &str) -> Option { + match value { + "active" => Some(TopicStatus::Active), + "paused" => Some(TopicStatus::Paused), + "archived" => Some(TopicStatus::Archived), + _ => None, + } + } +} + +impl std::fmt::Display for TopicStatus { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.as_str()) + } +} + +/// Structured criteria for querying memories during topic synthesis. +/// Maps to `SearchConfig` plus SQL-level filters. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TopicCriteria { + /// Semantic query for hybrid search (omit for pure metadata queries). + #[serde(default)] + pub query: Option, + /// Filter to specific memory types (empty = all types). + #[serde(default)] + pub memory_types: Vec, + /// Minimum importance threshold. + #[serde(default)] + pub min_importance: Option, + /// Only memories from these channels (empty = all channels). + #[serde(default)] + pub channel_ids: Vec, + /// Only memories newer than this duration ago (e.g. "30d", "7d"). + #[serde(default)] + pub max_age: Option, + /// Search mode override (defaults to Hybrid if query present, Recent otherwise). + #[serde(default)] + pub mode: Option, + /// Sort order for non-hybrid modes. + #[serde(default)] + pub sort_by: Option, + /// Max memories to gather before synthesis (default 30). + #[serde(default = "default_max_memories")] + pub max_memories: usize, +} + +fn default_max_memories() -> usize { + 30 +} + +impl Default for TopicCriteria { + fn default() -> Self { + Self { + query: None, + memory_types: Vec::new(), + min_importance: None, + channel_ids: Vec::new(), + max_age: None, + mode: None, + sort_by: None, + max_memories: default_max_memories(), + } + } +} + +impl TopicCriteria { + /// Build a `SearchConfig` from criteria fields. + pub fn to_search_config(&self) -> SearchConfig { + let mode = self.mode.unwrap_or_else(|| { + if self.query.is_some() { + SearchMode::Hybrid + } else if !self.memory_types.is_empty() { + SearchMode::Typed + } else { + SearchMode::Recent + } + }); + let sort_by = self.sort_by.unwrap_or(SearchSort::Recent); + let memory_type = self.memory_types.first().copied(); + + SearchConfig { + mode, + memory_type, + sort_by, + max_results: self.max_memories, + ..Default::default() + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Topic { + pub id: String, + pub agent_id: String, + pub title: String, + pub content: String, + pub criteria: TopicCriteria, + pub pin_ids: Vec, + pub channel_ids: Vec, + pub status: TopicStatus, + pub max_words: usize, + pub last_memory_at: Option, + pub last_synced_at: Option, + pub created_at: String, + pub updated_at: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TopicVersion { + pub id: String, + pub topic_id: String, + pub content: String, + pub memory_count: i64, + pub created_at: String, +} + +#[derive(Debug, Clone)] +pub struct TopicStore { + pool: SqlitePool, +} + +impl TopicStore { + pub fn new(pool: SqlitePool) -> Self { + Self { pool } + } + + pub async fn create(&self, topic: &Topic) -> Result<()> { + let criteria_json = + serde_json::to_string(&topic.criteria).context("failed to serialize topic criteria")?; + let pin_ids_json = + serde_json::to_string(&topic.pin_ids).context("failed to serialize topic pin_ids")?; + let channel_ids_json = serde_json::to_string(&topic.channel_ids) + .context("failed to serialize topic channel_ids")?; + + sqlx::query( + r#" + INSERT INTO topics ( + id, agent_id, title, content, criteria, pin_ids, channel_ids, + status, max_words, last_memory_at, last_synced_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + "#, + ) + .bind(&topic.id) + .bind(&topic.agent_id) + .bind(&topic.title) + .bind(&topic.content) + .bind(&criteria_json) + .bind(&pin_ids_json) + .bind(&channel_ids_json) + .bind(topic.status.as_str()) + .bind(topic.max_words as i64) + .bind(&topic.last_memory_at) + .bind(&topic.last_synced_at) + .execute(&self.pool) + .await + .context("failed to insert topic")?; + + Ok(()) + } + + pub async fn get(&self, id: &str) -> Result> { + let row = sqlx::query( + "SELECT id, agent_id, title, content, criteria, pin_ids, channel_ids, \ + status, max_words, last_memory_at, last_synced_at, created_at, updated_at \ + FROM topics WHERE id = ?", + ) + .bind(id) + .fetch_optional(&self.pool) + .await + .context("failed to fetch topic")?; + + row.map(topic_from_row).transpose() + } + + pub async fn list(&self, agent_id: &str) -> Result> { + let rows = sqlx::query( + "SELECT id, agent_id, title, content, criteria, pin_ids, channel_ids, \ + status, max_words, last_memory_at, last_synced_at, created_at, updated_at \ + FROM topics WHERE agent_id = ? ORDER BY created_at DESC", + ) + .bind(agent_id) + .fetch_all(&self.pool) + .await + .context("failed to list topics")?; + + rows.into_iter().map(topic_from_row).collect() + } + + pub async fn list_active(&self, agent_id: &str) -> Result> { + let rows = sqlx::query( + "SELECT id, agent_id, title, content, criteria, pin_ids, channel_ids, \ + status, max_words, last_memory_at, last_synced_at, created_at, updated_at \ + FROM topics WHERE agent_id = ? AND status = 'active' ORDER BY created_at DESC", + ) + .bind(agent_id) + .fetch_all(&self.pool) + .await + .context("failed to list active topics")?; + + rows.into_iter().map(topic_from_row).collect() + } + + pub async fn update(&self, topic: &Topic) -> Result<()> { + let criteria_json = + serde_json::to_string(&topic.criteria).context("failed to serialize topic criteria")?; + let pin_ids_json = + serde_json::to_string(&topic.pin_ids).context("failed to serialize topic pin_ids")?; + let channel_ids_json = serde_json::to_string(&topic.channel_ids) + .context("failed to serialize topic channel_ids")?; + + sqlx::query( + r#" + UPDATE topics SET + title = ?, content = ?, criteria = ?, pin_ids = ?, channel_ids = ?, + status = ?, max_words = ?, last_memory_at = ?, last_synced_at = ?, + updated_at = datetime('now') + WHERE id = ? + "#, + ) + .bind(&topic.title) + .bind(&topic.content) + .bind(&criteria_json) + .bind(&pin_ids_json) + .bind(&channel_ids_json) + .bind(topic.status.as_str()) + .bind(topic.max_words as i64) + .bind(&topic.last_memory_at) + .bind(&topic.last_synced_at) + .bind(&topic.id) + .execute(&self.pool) + .await + .context("failed to update topic")?; + + Ok(()) + } + + pub async fn delete(&self, id: &str) -> Result { + let result = sqlx::query("DELETE FROM topics WHERE id = ?") + .bind(id) + .execute(&self.pool) + .await + .context("failed to delete topic")?; + + Ok(result.rows_affected() > 0) + } + + /// Load all active topics that are assigned to a specific channel. + pub async fn get_for_channel(&self, agent_id: &str, channel_id: &str) -> Result> { + // channel_ids is a JSON array stored as TEXT. Use json_each to match. + let rows = sqlx::query( + "SELECT t.id, t.agent_id, t.title, t.content, t.criteria, t.pin_ids, t.channel_ids, \ + t.status, t.max_words, t.last_memory_at, t.last_synced_at, t.created_at, t.updated_at \ + FROM topics t, json_each(t.channel_ids) j \ + WHERE t.agent_id = ? AND t.status = 'active' AND j.value = ? \ + ORDER BY t.created_at DESC", + ) + .bind(agent_id) + .bind(channel_id) + .fetch_all(&self.pool) + .await + .context("failed to fetch topics for channel")?; + + rows.into_iter().map(topic_from_row).collect() + } + + pub async fn save_version(&self, version: &TopicVersion) -> Result<()> { + sqlx::query( + "INSERT INTO topic_versions (id, topic_id, content, memory_count) VALUES (?, ?, ?, ?)", + ) + .bind(&version.id) + .bind(&version.topic_id) + .bind(&version.content) + .bind(version.memory_count) + .execute(&self.pool) + .await + .context("failed to save topic version")?; + + Ok(()) + } + + pub async fn get_versions(&self, topic_id: &str, limit: i64) -> Result> { + let rows = sqlx::query( + "SELECT id, topic_id, content, memory_count, created_at \ + FROM topic_versions WHERE topic_id = ? ORDER BY created_at DESC LIMIT ?", + ) + .bind(topic_id) + .bind(limit.clamp(1, 100)) + .fetch_all(&self.pool) + .await + .context("failed to fetch topic versions")?; + + rows.into_iter().map(version_from_row).collect() + } + + /// Clear `last_synced_at` on all active topics for an agent, forcing + /// the sync loop to re-evaluate and re-synthesize them. + pub async fn mark_all_stale(&self, agent_id: &str) -> Result<()> { + sqlx::query( + "UPDATE topics SET last_synced_at = NULL WHERE agent_id = ? AND status = 'active'", + ) + .bind(agent_id) + .execute(&self.pool) + .await + .context("failed to mark topics stale")?; + + Ok(()) + } +} + +fn parse_json_string_array(value: &str) -> Vec { + serde_json::from_str(value).unwrap_or_default() +} + +fn parse_criteria(value: &str) -> TopicCriteria { + serde_json::from_str(value).unwrap_or_default() +} + +fn topic_from_row(row: sqlx::sqlite::SqliteRow) -> Result { + let status_value: String = row + .try_get("status") + .context("failed to read topic status")?; + let criteria_value: String = row.try_get("criteria").unwrap_or_else(|_| "{}".to_string()); + let pin_ids_value: String = row.try_get("pin_ids").unwrap_or_else(|_| "[]".to_string()); + let channel_ids_value: String = row + .try_get("channel_ids") + .unwrap_or_else(|_| "[]".to_string()); + + let status = TopicStatus::parse(&status_value) + .with_context(|| format!("invalid topic status in database: {status_value}"))?; + + Ok(Topic { + id: row.try_get("id").context("failed to read topic id")?, + agent_id: row + .try_get("agent_id") + .context("failed to read topic agent_id")?, + title: row.try_get("title").context("failed to read topic title")?, + content: row + .try_get("content") + .context("failed to read topic content")?, + criteria: parse_criteria(&criteria_value), + pin_ids: parse_json_string_array(&pin_ids_value), + channel_ids: parse_json_string_array(&channel_ids_value), + status, + max_words: row + .try_get::("max_words") + .map(|value| value as usize) + .unwrap_or(1500), + last_memory_at: row.try_get("last_memory_at").ok(), + last_synced_at: row.try_get("last_synced_at").ok(), + created_at: row + .try_get::("created_at") + .map(|value| value.and_utc().to_rfc3339()) + .context("failed to read topic created_at")?, + updated_at: row + .try_get::("updated_at") + .map(|value| value.and_utc().to_rfc3339()) + .context("failed to read topic updated_at")?, + }) +} + +fn version_from_row(row: sqlx::sqlite::SqliteRow) -> Result { + Ok(TopicVersion { + id: row + .try_get("id") + .context("failed to read topic version id")?, + topic_id: row + .try_get("topic_id") + .context("failed to read topic version topic_id")?, + content: row + .try_get("content") + .context("failed to read topic version content")?, + memory_count: row + .try_get("memory_count") + .context("failed to read topic version memory_count")?, + created_at: row + .try_get::("created_at") + .map(|value| value.and_utc().to_rfc3339()) + .context("failed to read topic version created_at")?, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use sqlx::sqlite::SqlitePoolOptions; + + async fn setup_store() -> TopicStore { + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .expect("in-memory sqlite should connect"); + + sqlx::query( + r#" + CREATE TABLE topics ( + id TEXT PRIMARY KEY, + agent_id TEXT NOT NULL, + title TEXT NOT NULL, + content TEXT NOT NULL DEFAULT '', + criteria TEXT NOT NULL, + pin_ids TEXT NOT NULL DEFAULT '[]', + channel_ids TEXT NOT NULL DEFAULT '[]', + status TEXT NOT NULL DEFAULT 'active', + max_words INTEGER NOT NULL DEFAULT 1500, + last_memory_at TEXT, + last_synced_at TEXT, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP + ) + "#, + ) + .execute(&pool) + .await + .expect("topics schema should be created"); + + sqlx::query( + r#" + CREATE TABLE topic_versions ( + id TEXT PRIMARY KEY, + topic_id TEXT NOT NULL, + content TEXT NOT NULL, + memory_count INTEGER NOT NULL, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY (topic_id) REFERENCES topics(id) ON DELETE CASCADE + ) + "#, + ) + .execute(&pool) + .await + .expect("topic_versions schema should be created"); + + TopicStore::new(pool) + } + + fn make_topic(agent_id: &str, title: &str) -> Topic { + Topic { + id: uuid::Uuid::new_v4().to_string(), + agent_id: agent_id.to_string(), + title: title.to_string(), + content: String::new(), + criteria: TopicCriteria::default(), + pin_ids: Vec::new(), + channel_ids: Vec::new(), + status: TopicStatus::Active, + max_words: 1500, + last_memory_at: None, + last_synced_at: None, + created_at: chrono::Utc::now().to_rfc3339(), + updated_at: chrono::Utc::now().to_rfc3339(), + } + } + + #[tokio::test] + async fn create_and_get() { + let store = setup_store().await; + let topic = make_topic("agent-1", "Test Topic"); + store.create(&topic).await.expect("create should succeed"); + + let loaded = store + .get(&topic.id) + .await + .expect("get should succeed") + .expect("topic should exist"); + assert_eq!(loaded.title, "Test Topic"); + assert_eq!(loaded.status, TopicStatus::Active); + } + + #[tokio::test] + async fn list_and_list_active() { + let store = setup_store().await; + + let mut active = make_topic("agent-1", "Active Topic"); + active.status = TopicStatus::Active; + store.create(&active).await.expect("create should succeed"); + + let mut paused = make_topic("agent-1", "Paused Topic"); + paused.status = TopicStatus::Paused; + store.create(&paused).await.expect("create should succeed"); + + let all = store.list("agent-1").await.expect("list should succeed"); + assert_eq!(all.len(), 2); + + let active_only = store + .list_active("agent-1") + .await + .expect("list_active should succeed"); + assert_eq!(active_only.len(), 1); + assert_eq!(active_only[0].title, "Active Topic"); + } + + #[tokio::test] + async fn update_topic() { + let store = setup_store().await; + let mut topic = make_topic("agent-1", "Original"); + store.create(&topic).await.expect("create should succeed"); + + topic.title = "Updated".to_string(); + topic.content = "Some content".to_string(); + store.update(&topic).await.expect("update should succeed"); + + let loaded = store + .get(&topic.id) + .await + .expect("get should succeed") + .expect("topic should exist"); + assert_eq!(loaded.title, "Updated"); + assert_eq!(loaded.content, "Some content"); + } + + #[tokio::test] + async fn delete_topic() { + let store = setup_store().await; + let topic = make_topic("agent-1", "Delete Me"); + store.create(&topic).await.expect("create should succeed"); + + let deleted = store + .delete(&topic.id) + .await + .expect("delete should succeed"); + assert!(deleted); + + let loaded = store.get(&topic.id).await.expect("get should succeed"); + assert!(loaded.is_none()); + } + + #[tokio::test] + async fn get_for_channel() { + let store = setup_store().await; + + let mut topic = make_topic("agent-1", "Channel Topic"); + topic.channel_ids = vec!["chan-1".to_string(), "chan-2".to_string()]; + store.create(&topic).await.expect("create should succeed"); + + let mut other = make_topic("agent-1", "Other Topic"); + other.channel_ids = vec!["chan-3".to_string()]; + store.create(&other).await.expect("create should succeed"); + + let results = store + .get_for_channel("agent-1", "chan-1") + .await + .expect("get_for_channel should succeed"); + assert_eq!(results.len(), 1); + assert_eq!(results[0].title, "Channel Topic"); + + let empty = store + .get_for_channel("agent-1", "chan-99") + .await + .expect("get_for_channel should succeed"); + assert!(empty.is_empty()); + } + + #[tokio::test] + async fn version_history() { + let store = setup_store().await; + let topic = make_topic("agent-1", "Versioned"); + store.create(&topic).await.expect("create should succeed"); + + let version = TopicVersion { + id: uuid::Uuid::new_v4().to_string(), + topic_id: topic.id.clone(), + content: "Version 1 content".to_string(), + memory_count: 15, + created_at: chrono::Utc::now().to_rfc3339(), + }; + store + .save_version(&version) + .await + .expect("save_version should succeed"); + + let versions = store + .get_versions(&topic.id, 10) + .await + .expect("get_versions should succeed"); + assert_eq!(versions.len(), 1); + assert_eq!(versions[0].memory_count, 15); + } +} diff --git a/tests/bulletin.rs b/tests/bulletin.rs index c6f43529c..343e713bf 100644 --- a/tests/bulletin.rs +++ b/tests/bulletin.rs @@ -112,6 +112,7 @@ async fn bootstrap_deps() -> anyhow::Result { llm_manager, mcp_manager, task_store, + topic_store: Arc::new(spacebot::topics::TopicStore::new(db.sqlite.clone())), cron_tool: None, runtime_config, event_tx, diff --git a/tests/context_dump.rs b/tests/context_dump.rs index df77d7807..143f11dfe 100644 --- a/tests/context_dump.rs +++ b/tests/context_dump.rs @@ -111,6 +111,7 @@ async fn bootstrap_deps() -> anyhow::Result<(spacebot::AgentDeps, spacebot::conf llm_manager, mcp_manager, task_store, + topic_store: Arc::new(spacebot::topics::TopicStore::new(db.sqlite.clone())), cron_tool: None, runtime_config, event_tx, From 65a723b35ceb068eea78e6bcaa7e33903c0eb61b Mon Sep 17 00:00:00 2001 From: James Pine Date: Mon, 2 Mar 2026 16:35:40 -0800 Subject: [PATCH 2/5] refactor: topics as recall layer, bulletin always injected MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Topics are no longer assigned to channels — the bulletin remains the universal awareness floor for all channels/branches. Topics exist as standalone context documents that workers and branches pull in explicitly via topic_ids on spawn_worker. - Always inject memory bulletin into channel/cortex-chat system prompts - List available topics (id + title) in the status block so the LLM can reference them when spawning workers - Remove channel_ids from Topic struct, store, migration, and API - Remove get_for_channel query (no longer needed) - Fix collapsible-if clippy lints in channel_dispatch and cortex --- migrations/20260302000001_topics.sql | 1 - src/agent/channel.rs | 49 ++++++++----------- src/agent/channel_dispatch.rs | 9 ++-- src/agent/cortex.rs | 18 ++++--- src/agent/cortex_chat.rs | 19 +------- src/api/topics.rs | 8 ---- src/topics/store.rs | 70 +++------------------------- 7 files changed, 38 insertions(+), 136 deletions(-) diff --git a/migrations/20260302000001_topics.sql b/migrations/20260302000001_topics.sql index 399d9dd2f..853ddcaae 100644 --- a/migrations/20260302000001_topics.sql +++ b/migrations/20260302000001_topics.sql @@ -7,7 +7,6 @@ CREATE TABLE IF NOT EXISTS topics ( content TEXT NOT NULL DEFAULT '', criteria TEXT NOT NULL, pin_ids TEXT NOT NULL DEFAULT '[]', - channel_ids TEXT NOT NULL DEFAULT '[]', status TEXT NOT NULL DEFAULT 'active', max_words INTEGER NOT NULL DEFAULT 1500, last_memory_at TEXT, diff --git a/src/agent/channel.rs b/src/agent/channel.rs index 1897b9950..95a9d1946 100644 --- a/src/agent/channel.rs +++ b/src/agent/channel.rs @@ -675,22 +675,25 @@ impl Channel { let identity_context = rc.identity.load().render(); - // Use channel-specific topics if assigned, otherwise fall back to the global bulletin. - let channel_topics = self + let memory_bulletin = rc.memory_bulletin.load().to_string(); + + // List all active topics so the LLM can reference topic IDs when spawning workers. + let all_active_topics = self .deps .topic_store - .get_for_channel(&self.deps.agent_id, &self.id) + .list_active(&self.deps.agent_id) .await .unwrap_or_default(); - let memory_bulletin = if channel_topics.is_empty() { - rc.memory_bulletin.load().to_string() + let available_topics = if all_active_topics.is_empty() { + String::new() } else { - channel_topics - .iter() - .filter(|topic| !topic.content.is_empty()) - .map(|topic| format!("### {}\n\n{}", topic.title, topic.content)) - .collect::>() - .join("\n\n") + let mut section = String::from( + "\n## Available Topics\nYou can pass these topic IDs to `spawn_worker` via the `topic_ids` parameter for extra context:\n", + ); + for topic in &all_active_topics { + section.push_str(&format!("- `{}`: {}\n", topic.id, topic.title)); + } + section }; let skills = rc.skills.load(); @@ -710,7 +713,9 @@ impl Channel { let current_time_line = temporal_context.current_time_line(); let status_text = { let status = self.state.status_block.read().await; - status.render_with_time_context(Some(¤t_time_line)) + let mut rendered = status.render_with_time_context(Some(¤t_time_line)); + rendered.push_str(&available_topics); + rendered }; // Render coalesce hint @@ -1021,23 +1026,7 @@ impl Channel { let identity_context = rc.identity.load().render(); - // Use channel-specific topics if assigned, otherwise fall back to the global bulletin. - let channel_topics = self - .deps - .topic_store - .get_for_channel(&self.deps.agent_id, &self.id) - .await - .unwrap_or_default(); - let memory_bulletin = if channel_topics.is_empty() { - rc.memory_bulletin.load().to_string() - } else { - channel_topics - .iter() - .filter(|topic| !topic.content.is_empty()) - .map(|topic| format!("### {}\n\n{}", topic.title, topic.content)) - .collect::>() - .join("\n\n") - }; + let memory_bulletin = rc.memory_bulletin.load().to_string(); // List all active topics so the LLM can reference topic IDs when spawning workers. let all_active_topics = self @@ -1050,7 +1039,7 @@ impl Channel { String::new() } else { let mut section = String::from( - "\n## Available Topics\nYou can pass these topic IDs to `spawn_worker` via the `topic_ids` parameter:\n", + "\n## Available Topics\nYou can pass these topic IDs to `spawn_worker` via the `topic_ids` parameter for extra context:\n", ); for topic in &all_active_topics { section.push_str(&format!("- `{}`: {}\n", topic.id, topic.title)); diff --git a/src/agent/channel_dispatch.rs b/src/agent/channel_dispatch.rs index 36fdcecd2..313bbdae2 100644 --- a/src/agent/channel_dispatch.rs +++ b/src/agent/channel_dispatch.rs @@ -257,11 +257,10 @@ pub async fn spawn_worker_from_state( if !topic_ids.is_empty() { let mut topic_context = String::from("## Topic Context\n\n"); for topic_id in topic_ids { - if let Ok(Some(topic)) = state.deps.topic_store.get(topic_id).await { - if !topic.content.is_empty() { - topic_context - .push_str(&format!("### {}\n\n{}\n\n", topic.title, topic.content)); - } + if let Ok(Some(topic)) = state.deps.topic_store.get(topic_id).await + && !topic.content.is_empty() + { + topic_context.push_str(&format!("### {}\n\n{}\n\n", topic.title, topic.content)); } } if topic_context.len() > "## Topic Context\n\n".len() { diff --git a/src/agent/cortex.rs b/src/agent/cortex.rs index c41f12d6d..20986f91a 100644 --- a/src/agent/cortex.rs +++ b/src/agent/cortex.rs @@ -375,15 +375,14 @@ impl Cortex { // When a memory is saved, mark all active topics as stale so the // topic sync loop re-synthesizes them on its next pass. - if is_memory_saved { - if let Err(error) = self + if is_memory_saved + && let Err(error) = self .deps .topic_store .mark_all_stale(&self.deps.agent_id) .await - { - tracing::warn!(%error, "failed to mark topics stale after memory save"); - } + { + tracing::warn!(%error, "failed to mark topics stale after memory save"); } } @@ -1222,11 +1221,10 @@ async fn pickup_one_ready_task(deps: &AgentDeps, logger: &CortexLogger) -> anyho if !topic_ids.is_empty() { let mut topic_content = String::new(); for topic_id in &topic_ids { - if let Ok(Some(topic)) = deps.topic_store.get(topic_id).await { - if !topic.content.is_empty() { - topic_content - .push_str(&format!("\n## {}\n{}\n", topic.title, topic.content)); - } + if let Ok(Some(topic)) = deps.topic_store.get(topic_id).await + && !topic.content.is_empty() + { + topic_content.push_str(&format!("\n## {}\n{}\n", topic.title, topic.content)); } } if !topic_content.is_empty() { diff --git a/src/agent/cortex_chat.rs b/src/agent/cortex_chat.rs index 41bb24c50..af7ccc40b 100644 --- a/src/agent/cortex_chat.rs +++ b/src/agent/cortex_chat.rs @@ -329,24 +329,7 @@ impl CortexChatSession { let identity_context = runtime_config.identity.load().render(); - // Compose all active topics for the cortex chat context, falling - // back to the global bulletin if no topics exist. - let all_topics = self - .deps - .topic_store - .list_active(&self.deps.agent_id) - .await - .unwrap_or_default(); - let context_content = if all_topics.is_empty() { - runtime_config.memory_bulletin.load().to_string() - } else { - all_topics - .iter() - .filter(|topic| !topic.content.is_empty()) - .map(|topic| format!("### {}\n\n{}", topic.title, topic.content)) - .collect::>() - .join("\n\n") - }; + let context_content = runtime_config.memory_bulletin.load().to_string(); let browser_enabled = runtime_config.browser_config.load().enabled; let web_search_enabled = runtime_config.brave_search_key.load().is_some(); diff --git a/src/api/topics.rs b/src/api/topics.rs index 29ddeb6b9..d1a6552fe 100644 --- a/src/api/topics.rs +++ b/src/api/topics.rs @@ -27,8 +27,6 @@ pub(super) struct CreateTopicRequest { #[serde(default)] criteria: Option, #[serde(default)] - channel_ids: Vec, - #[serde(default)] pin_ids: Vec, #[serde(default)] status: Option, @@ -44,8 +42,6 @@ pub(super) struct UpdateTopicRequest { #[serde(default)] criteria: Option, #[serde(default)] - channel_ids: Option>, - #[serde(default)] pin_ids: Option>, #[serde(default)] status: Option, @@ -160,7 +156,6 @@ pub(super) async fn create_topic( content: String::new(), criteria: request.criteria.unwrap_or_default(), pin_ids: request.pin_ids, - channel_ids: request.channel_ids, status, max_words: request.max_words.unwrap_or(1500), last_memory_at: None, @@ -200,9 +195,6 @@ pub(super) async fn update_topic( if let Some(criteria) = request.criteria { topic.criteria = criteria; } - if let Some(channel_ids) = request.channel_ids { - topic.channel_ids = channel_ids; - } if let Some(pin_ids) = request.pin_ids { topic.pin_ids = pin_ids; } diff --git a/src/topics/store.rs b/src/topics/store.rs index 417eec3f7..f7f2980d8 100644 --- a/src/topics/store.rs +++ b/src/topics/store.rs @@ -122,7 +122,6 @@ pub struct Topic { pub content: String, pub criteria: TopicCriteria, pub pin_ids: Vec, - pub channel_ids: Vec, pub status: TopicStatus, pub max_words: usize, pub last_memory_at: Option, @@ -155,16 +154,14 @@ impl TopicStore { serde_json::to_string(&topic.criteria).context("failed to serialize topic criteria")?; let pin_ids_json = serde_json::to_string(&topic.pin_ids).context("failed to serialize topic pin_ids")?; - let channel_ids_json = serde_json::to_string(&topic.channel_ids) - .context("failed to serialize topic channel_ids")?; sqlx::query( r#" INSERT INTO topics ( - id, agent_id, title, content, criteria, pin_ids, channel_ids, + id, agent_id, title, content, criteria, pin_ids, status, max_words, last_memory_at, last_synced_at ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) "#, ) .bind(&topic.id) @@ -173,7 +170,6 @@ impl TopicStore { .bind(&topic.content) .bind(&criteria_json) .bind(&pin_ids_json) - .bind(&channel_ids_json) .bind(topic.status.as_str()) .bind(topic.max_words as i64) .bind(&topic.last_memory_at) @@ -187,7 +183,7 @@ impl TopicStore { pub async fn get(&self, id: &str) -> Result> { let row = sqlx::query( - "SELECT id, agent_id, title, content, criteria, pin_ids, channel_ids, \ + "SELECT id, agent_id, title, content, criteria, pin_ids, \ status, max_words, last_memory_at, last_synced_at, created_at, updated_at \ FROM topics WHERE id = ?", ) @@ -201,7 +197,7 @@ impl TopicStore { pub async fn list(&self, agent_id: &str) -> Result> { let rows = sqlx::query( - "SELECT id, agent_id, title, content, criteria, pin_ids, channel_ids, \ + "SELECT id, agent_id, title, content, criteria, pin_ids, \ status, max_words, last_memory_at, last_synced_at, created_at, updated_at \ FROM topics WHERE agent_id = ? ORDER BY created_at DESC", ) @@ -215,7 +211,7 @@ impl TopicStore { pub async fn list_active(&self, agent_id: &str) -> Result> { let rows = sqlx::query( - "SELECT id, agent_id, title, content, criteria, pin_ids, channel_ids, \ + "SELECT id, agent_id, title, content, criteria, pin_ids, \ status, max_words, last_memory_at, last_synced_at, created_at, updated_at \ FROM topics WHERE agent_id = ? AND status = 'active' ORDER BY created_at DESC", ) @@ -232,13 +228,11 @@ impl TopicStore { serde_json::to_string(&topic.criteria).context("failed to serialize topic criteria")?; let pin_ids_json = serde_json::to_string(&topic.pin_ids).context("failed to serialize topic pin_ids")?; - let channel_ids_json = serde_json::to_string(&topic.channel_ids) - .context("failed to serialize topic channel_ids")?; sqlx::query( r#" UPDATE topics SET - title = ?, content = ?, criteria = ?, pin_ids = ?, channel_ids = ?, + title = ?, content = ?, criteria = ?, pin_ids = ?, status = ?, max_words = ?, last_memory_at = ?, last_synced_at = ?, updated_at = datetime('now') WHERE id = ? @@ -248,7 +242,6 @@ impl TopicStore { .bind(&topic.content) .bind(&criteria_json) .bind(&pin_ids_json) - .bind(&channel_ids_json) .bind(topic.status.as_str()) .bind(topic.max_words as i64) .bind(&topic.last_memory_at) @@ -271,25 +264,6 @@ impl TopicStore { Ok(result.rows_affected() > 0) } - /// Load all active topics that are assigned to a specific channel. - pub async fn get_for_channel(&self, agent_id: &str, channel_id: &str) -> Result> { - // channel_ids is a JSON array stored as TEXT. Use json_each to match. - let rows = sqlx::query( - "SELECT t.id, t.agent_id, t.title, t.content, t.criteria, t.pin_ids, t.channel_ids, \ - t.status, t.max_words, t.last_memory_at, t.last_synced_at, t.created_at, t.updated_at \ - FROM topics t, json_each(t.channel_ids) j \ - WHERE t.agent_id = ? AND t.status = 'active' AND j.value = ? \ - ORDER BY t.created_at DESC", - ) - .bind(agent_id) - .bind(channel_id) - .fetch_all(&self.pool) - .await - .context("failed to fetch topics for channel")?; - - rows.into_iter().map(topic_from_row).collect() - } - pub async fn save_version(&self, version: &TopicVersion) -> Result<()> { sqlx::query( "INSERT INTO topic_versions (id, topic_id, content, memory_count) VALUES (?, ?, ?, ?)", @@ -348,9 +322,6 @@ fn topic_from_row(row: sqlx::sqlite::SqliteRow) -> Result { .context("failed to read topic status")?; let criteria_value: String = row.try_get("criteria").unwrap_or_else(|_| "{}".to_string()); let pin_ids_value: String = row.try_get("pin_ids").unwrap_or_else(|_| "[]".to_string()); - let channel_ids_value: String = row - .try_get("channel_ids") - .unwrap_or_else(|_| "[]".to_string()); let status = TopicStatus::parse(&status_value) .with_context(|| format!("invalid topic status in database: {status_value}"))?; @@ -366,7 +337,6 @@ fn topic_from_row(row: sqlx::sqlite::SqliteRow) -> Result { .context("failed to read topic content")?, criteria: parse_criteria(&criteria_value), pin_ids: parse_json_string_array(&pin_ids_value), - channel_ids: parse_json_string_array(&channel_ids_value), status, max_words: row .try_get::("max_words") @@ -427,7 +397,6 @@ mod tests { content TEXT NOT NULL DEFAULT '', criteria TEXT NOT NULL, pin_ids TEXT NOT NULL DEFAULT '[]', - channel_ids TEXT NOT NULL DEFAULT '[]', status TEXT NOT NULL DEFAULT 'active', max_words INTEGER NOT NULL DEFAULT 1500, last_memory_at TEXT, @@ -468,7 +437,6 @@ mod tests { content: String::new(), criteria: TopicCriteria::default(), pin_ids: Vec::new(), - channel_ids: Vec::new(), status: TopicStatus::Active, max_words: 1500, last_memory_at: None, @@ -551,32 +519,6 @@ mod tests { assert!(loaded.is_none()); } - #[tokio::test] - async fn get_for_channel() { - let store = setup_store().await; - - let mut topic = make_topic("agent-1", "Channel Topic"); - topic.channel_ids = vec!["chan-1".to_string(), "chan-2".to_string()]; - store.create(&topic).await.expect("create should succeed"); - - let mut other = make_topic("agent-1", "Other Topic"); - other.channel_ids = vec!["chan-3".to_string()]; - store.create(&other).await.expect("create should succeed"); - - let results = store - .get_for_channel("agent-1", "chan-1") - .await - .expect("get_for_channel should succeed"); - assert_eq!(results.len(), 1); - assert_eq!(results[0].title, "Channel Topic"); - - let empty = store - .get_for_channel("agent-1", "chan-99") - .await - .expect("get_for_channel should succeed"); - assert!(empty.is_empty()); - } - #[tokio::test] async fn version_history() { let store = setup_store().await; From 8156d9951930d69bc991573ce833a594053da8ef Mon Sep 17 00:00:00 2001 From: James Pine Date: Mon, 2 Mar 2026 19:37:16 -0800 Subject: [PATCH 3/5] fix: remove channel_ids from topic frontend types and UI Removes channel assignment UI from create/edit topic dialogs and the channel count badges from topic cards/detail views. Fixes 'undefined is not an object' crash from accessing topic.channel_ids after the backend stopped sending it. --- interface/src/api/client.ts | 3 - interface/src/routes/AgentTopics.tsx | 109 +-------------------------- 2 files changed, 1 insertion(+), 111 deletions(-) diff --git a/interface/src/api/client.ts b/interface/src/api/client.ts index 9c8bc285e..fce3bbd52 100644 --- a/interface/src/api/client.ts +++ b/interface/src/api/client.ts @@ -969,7 +969,6 @@ export interface TopicItem { content: string; criteria: TopicCriteria; pin_ids: string[]; - channel_ids: string[]; status: TopicStatus; max_words: number; last_memory_at: string | null; @@ -1006,7 +1005,6 @@ export interface TopicVersionsResponse { export interface CreateTopicRequest { title: string; criteria?: TopicCriteria; - channel_ids?: string[]; pin_ids?: string[]; status?: TopicStatus; max_words?: number; @@ -1015,7 +1013,6 @@ export interface CreateTopicRequest { export interface UpdateTopicRequest { title?: string; criteria?: TopicCriteria; - channel_ids?: string[]; pin_ids?: string[]; status?: TopicStatus; max_words?: number; diff --git a/interface/src/routes/AgentTopics.tsx b/interface/src/routes/AgentTopics.tsx index 7bddb702b..b90946865 100644 --- a/interface/src/routes/AgentTopics.tsx +++ b/interface/src/routes/AgentTopics.tsx @@ -8,7 +8,6 @@ import { type CreateTopicRequest, type UpdateTopicRequest, type TopicCriteria, - type ChannelInfo, type MemoryItem, MEMORY_TYPES, type MemoryType, @@ -50,15 +49,6 @@ export function AgentTopics({ agentId }: { agentId: string }) { refetchInterval: 15_000, }); - const { data: channelsData } = useQuery({ - queryKey: ["channels"], - queryFn: api.channels, - }); - - const agentChannels = (channelsData?.channels ?? []).filter( - (c) => c.agent_id === agentId, - ); - const topics = data?.topics ?? []; const [createOpen, setCreateOpen] = useState(false); @@ -126,7 +116,6 @@ export function AgentTopics({ agentId }: { agentId: string }) { onClose={() => setCreateOpen(false)} onCreate={(request) => createMutation.mutate(request)} isPending={createMutation.isPending} - channels={agentChannels} /> ); @@ -171,7 +160,6 @@ export function AgentTopics({ agentId }: { agentId: string }) { onClose={() => setCreateOpen(false)} onCreate={(request) => createMutation.mutate(request)} isPending={createMutation.isPending} - channels={agentChannels} /> {/* Detail Dialog */} @@ -179,7 +167,6 @@ export function AgentTopics({ agentId }: { agentId: string }) { setSelectedTopicId(null)} onUpdate={(request) => updateMutation.mutate({ @@ -239,12 +226,6 @@ function TopicCard({ Synced {formatTimeAgo(topic.last_synced_at)} )} {!topic.last_synced_at && Never synced} - {topic.channel_ids.length > 0 && ( - - {topic.channel_ids.length} channel - {topic.channel_ids.length !== 1 ? "s" : ""} - - )} {wordCount > 0 && {wordCount} words} @@ -268,20 +249,17 @@ function CreateTopicDialog({ onClose, onCreate, isPending, - channels, }: { open: boolean; onClose: () => void; onCreate: (request: CreateTopicRequest) => void; isPending: boolean; - channels: ChannelInfo[]; }) { const [title, setTitle] = useState(""); const [query, setQuery] = useState(""); const [selectedTypes, setSelectedTypes] = useState([]); const [maxAge, setMaxAge] = useState(""); const [maxWords, setMaxWords] = useState(1500); - const [selectedChannels, setSelectedChannels] = useState([]); const handleSubmit = useCallback(() => { if (!title.trim()) return; @@ -292,7 +270,6 @@ function CreateTopicDialog({ onCreate({ title: title.trim(), criteria: Object.keys(criteria).length > 0 ? criteria : undefined, - channel_ids: selectedChannels.length > 0 ? selectedChannels : undefined, max_words: maxWords, }); setTitle(""); @@ -300,8 +277,7 @@ function CreateTopicDialog({ setSelectedTypes([]); setMaxAge(""); setMaxWords(1500); - setSelectedChannels([]); - }, [title, query, selectedTypes, maxAge, maxWords, selectedChannels, onCreate]); + }, [title, query, selectedTypes, maxAge, maxWords, onCreate]); return ( !v && onClose()}> @@ -358,38 +334,6 @@ function CreateTopicDialog({ ))} - {channels.length > 0 && ( -
- -
- {channels.map((channel) => ( - - ))} -
-
- )}
- {/* Channel Assignments */} - {channels.length > 0 && ( -
- -
- {channels.map((channel) => ( - - ))} -
-
- )} - {/* Semantic Query */}