merge: adopt history search catalog from #8217

Add the token-only FTS5 history projection, non-blocking persist observer,
paged Desktop history manager, and CLI doctor/reindex surface on top of the
hardened session-catalog foundation. Keep corruption-only projectiondb
quarantine and session-catalog lifecycle fixes from #8186 while adopting
history session pagination (schema v3).
This commit is contained in:
SivanCola committed 2026-08-11 01:49:38 +08:00
commit 608543afee
46 files changed
+3070 -113

No files matched your search

+20 -20
View File
@@ -614,6 +614,7 @@ func (a *App) startup(ctx context.Context) {
a.tabsRestored = make(chan struct{})
a.mu.Unlock()
go a.restoreOrBuildTabs()
a.registerHistoryIndexEvents()
a.startSessionCatalog(false)
a.goSafe("refreshBotRuntime", a.refreshBotRuntime)
a.goSafe("sendStartupPing", a.sendStartupPing)
@@ -995,9 +996,10 @@ func (a *App) shutdown(context.Context) {
// Run after controller teardown (and after its deferred lifecycle unlocks)
// so every accepted usage record reaches disk before a normal app exit.
defer func() {
flushCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
flushCtx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond)
defer cancel()
_ = stats.Flush(flushCtx, config.StatsDir())
_ = flushDesktopDerivedCatalogs(flushCtx)
}()
a.stopDeferredRebuildRetry()
a.stopHistoryIndexMigration()
@@ -3154,30 +3156,28 @@ func (a *App) ListSessionsForTab(tabID string) []SessionMeta {
}
func (a *App) listSessionsFromDir(dir, active string) []SessionMeta {
infos, err := agent.ListSessions(dir)
catalog := a.sessionCatalog.Load()
if catalog == nil {
return []SessionMeta{}
}
target := sessioncatalog.DirectoryTarget{Path: dir, Scope: "global"}
for _, candidate := range a.sessionCatalogTargets() {
if sameProjectRoot(candidate.Path, dir) {
target = candidate
break
}
}
records, err := listCatalogSessionsForDirectory(a.bootContext(), catalog, target, dir)
if err != nil {
return []SessionMeta{}
}
open := a.openSessionPaths(dir)
protectedDisplays := make(map[string]struct{}, len(open))
for path := range open {
if key := filepath.Base(path); store.IsSessionTranscriptName(key) {
protectedDisplays[key] = struct{}{}
}
}
_ = pruneSessionDisplays(dir, protectedDisplays)
_ = pruneSessionPlannerDisplays(dir, protectedDisplays)
titles := loadSessionTitles(dir)
channelRoutes := channelSessionRoutesForDir(dir)
out := make([]SessionMeta, 0, len(infos))
for _, s := range infos {
_, isOpen := open[s.Path]
title := strings.TrimSpace(s.CustomTitle)
if title == "" {
title = titles[filepath.Base(s.Path)]
}
meta := sessionMetaFromInfo(s, title, s.Path == active, isOpen, 0, dir)
if route, ok := channelRoutes[sessionRuntimeKey(s.Path)]; ok {
out := make([]SessionMeta, 0, len(records))
for _, record := range records {
_, isOpen := open[record.Path]
meta := sessionMetaFromCatalog(record, record.Path == active, isOpen)
if route, ok := channelRoutes[sessionRuntimeKey(record.Path)]; ok {
applyChannelSessionRoute(&meta, route)
}
out = append(out, meta)
+5 -5
View File
@@ -4404,8 +4404,8 @@ func TestListSessionsUsesPinnedSessionOwnerBeforeStaleRuntimeDir(t *testing.T) {
app.tabs = map[string]*WorkspaceTab{tab.ID: tab}
app.tabOrder = []string{tab.ID}
app.activeTabID = tab.ID
installSessionCatalogForTest(t, app, sessionDirA, "project", projectA)
t.Cleanup(oldCtrl.Close)
sessions := app.ListSessions()
if len(sessions) == 0 {
t.Fatal("ListSessions() returned no sessions")
@@ -6692,7 +6692,7 @@ func TestDeleteSessionCancelsInactiveOpenRuntime(t *testing.T) {
tabOrder: []string{"active", "inactive"},
activeTabID: "active",
}
installSessionCatalogForTest(t, app, dir, "global", "")
if err := app.DeleteSession(filepath.Base(inactivePath)); err != nil {
t.Fatalf("DeleteSession(inactive open basename): %v", err)
}
@@ -6918,7 +6918,6 @@ func TestRestoreSessionRejectsDestroyingSession(t *testing.T) {
func TestDesktopSessionAPIsUseControllerSessionDir(t *testing.T) {
isolateDesktopUserDirs(t)
dirA := filepath.Join(t.TempDir(), "workspace-a-sessions")
dirB := filepath.Join(t.TempDir(), "workspace-b-sessions")
if err := os.MkdirAll(dirA, 0o755); err != nil {
@@ -6939,7 +6938,7 @@ func TestDesktopSessionAPIsUseControllerSessionDir(t *testing.T) {
app := NewApp()
app.setTestCtrl(control.New(control.Options{SessionDir: dirA, SessionPath: pathA, Label: "test"}), "")
defer app.activeCtrl().Close()
installSessionCatalogForTest(t, app, dirA, "global", "")
sessions := app.ListSessions()
if len(sessions) != 1 || sessions[0].Path != pathA || sessions[0].TurnsState != "unknown" ||
!strings.Contains(sessions[0].Preview, "being indexed") {
@@ -6955,6 +6954,7 @@ func TestDesktopSessionAPIsUseControllerSessionDir(t *testing.T) {
if meta.CustomTitle != "A title" {
t.Fatalf("custom title should be written to branch meta, got %q", meta.CustomTitle)
}
reconcileSessionCatalogForTest(t, app, dirA, "global", "")
sessions = app.ListSessions()
if len(sessions) != 1 || sessions[0].Title != "A title" {
t.Fatalf("ListSessions should return custom title from branch meta, got %+v", sessions)
@@ -6992,7 +6992,7 @@ func TestListSessionsMarksAutoBotSessionAsChannel(t *testing.T) {
app := NewApp()
app.setTestCtrl(control.New(control.Options{SessionDir: dir, SessionPath: filepath.Join(dir, "active.jsonl"), Label: "test"}), "")
defer app.activeCtrl().Close()
installSessionCatalogForTest(t, app, dir, "global", "")
sessions := app.ListSessions()
if len(sessions) != 1 {
t.Fatalf("ListSessions len = %d, want 1: %+v", len(sessions), sessions)
@@ -80,8 +80,8 @@ const rawInitialBytes = [...initialJS, ...initialCSS, ...appShellCSS]
.reduce((total, path) => total + statSync(path).size, 0);
// Native Web Animations and frame-batched scrolling avoid an eager animation
// runtime. Goal request observability plus transcript scroll arbitration,
// logical selection state/DOM adapters, measurement invalidation, and the
// catalog project shell add small always-available contracts; keep the raw
// allowance tight while gzip stays flat.
// selection lifecycle fencing, measurement invalidation, catalog project shell,
// and indexed-history bridge add small always-available contracts. Keep the raw
// allowance ratcheted while gzip stays flat.
assertBudget("initial raw JavaScript and CSS", rawInitialBytes, 2_240 * 1024);
assertBudget("largest initial JavaScript chunk raw", largestInitialJSRaw, 1_000 * 1024);
@@ -0,0 +1,30 @@
export function HistoryFilterSelect({
label,
options,
value,
onChange,
}: {
label: string;
options: { id: string; label: string; count: number }[];
value: string;
onChange: (next: string) => void;
}) {
const visibleOptions = options.filter((option) => option.id === "all" || option.id === value || option.count > 0);
return (
<div className="history-filter" role="group" aria-label={label}>
{visibleOptions.map((option) => (
<button
key={option.id}
type="button"
className={`history-filter__pill${value === option.id ? " history-filter__pill--on" : ""}`}
aria-pressed={value === option.id}
disabled={option.id !== "all" && option.count === 0}
onClick={() => onChange(option.id)}
>
{option.label}
<span className="history-filter__count">{option.count}</span>
</button>
))}
</div>
);
}
@@ -1,14 +1,17 @@
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import type { MouseEvent as ReactMouseEvent } from "react";
import { Archive, Pencil, Search, Trash2, RotateCcw } from "lucide-react";
import { app } from "../lib/bridge";
import { t, useT } from "../lib/i18n";
import { historySessionDisplayTitle, sessionActivityTime } from "../lib/session";
import type { HistoryMessage, SessionMeta } from "../lib/types";
import type { HistoryMessage, HistorySearchContextLine, HistorySearchHit, SessionMeta } from "../lib/types";
import { historyMessagesToItems, type Item } from "../lib/useController";
import { useHistoryCatalog } from "../lib/useHistoryCatalog";
import { Transcript } from "./Transcript";
import { ContextMenu, contextMenuPointFromEvent, type ContextMenuItem, type ContextMenuPoint } from "./ContextMenu";
import { useDeferredClose } from "../lib/useMountTransition";
import { ModalCloseButton } from "./ModalCloseButton";
import { HistoryFilterSelect } from "./HistoryFilterSelect";
type HistoryScopeFilter = "all" | "project" | "global";
type HistoryStatusFilter = "all" | "current" | "open";
@@ -19,7 +22,7 @@ type HistoryDateFilter = "all" | "today" | "yesterday" | "older";
// rename, or delete the selected session.
export function HistoryPanel({
kind = "history",
sessions,
sessions: suppliedSessions,
running,
onResume,
onPreview,
@@ -56,6 +59,10 @@ export function HistoryPanel({
const [scopeFilter, setScopeFilter] = useState<HistoryScopeFilter>("all");
const [statusFilter, setStatusFilter] = useState<HistoryStatusFilter>("all");
const [dateFilter, setDateFilter] = useState<HistoryDateFilter>("all");
const [searchContext, setSearchContext] = useState<{ hit: HistorySearchHit; lines: HistorySearchContextLine[]; loading: boolean } | null>(null);
const { sessions, nextCursor, partial: catalogPartial, progress: catalogProgress, searchHits, loadMore } = useHistoryCatalog({
isTrash, suppliedSessions, scope: scopeFilter, status: statusFilter, timeFilter: dateFilter, query,
});
const [menuSession, setMenuSession] = useState<SessionMeta | null>(null);
const [menuPoint, setMenuPoint] = useState<ContextMenuPoint | null>(null);
const [blankMenuPoint, setBlankMenuPoint] = useState<ContextMenuPoint | null>(null);
@@ -71,6 +78,14 @@ export function HistoryPanel({
} | null>(null);
const previewSeq = useRef(0);
const loadSearchContext = useCallback(async (hit: HistorySearchHit) => {
const seq = ++previewSeq.current;
setPreview(null);
setSearchContext({ hit, lines: [], loading: true });
const lines = await app.GetHistorySearchContext({ sessionPath: hit.sessionPath, messageIndex: hit.messageIndex, before: 2, after: 2 }).catch(() => []);
if (seq === previewSeq.current) setSearchContext({ hit, lines, loading: false });
}, []);
const startRename = (s: SessionMeta) => {
if (running) return;
setEditing(s.path);
@@ -84,6 +99,7 @@ export function HistoryPanel({
const loadPreview = useCallback(
async (s: SessionMeta) => {
const seq = ++previewSeq.current;
setSearchContext(null);
setEditing(null);
setPreview({
path: s.path,
@@ -123,22 +139,24 @@ export function HistoryPanel({
}, [isTrash, sessions]);
useEffect(() => {
if (!isTrash) return;
if (scopeFilter === "project" && scopeCounts.project === 0) setScopeFilter("all");
if (scopeFilter === "global" && scopeCounts.global === 0) setScopeFilter("all");
}, [scopeCounts.global, scopeCounts.project, scopeFilter]);
}, [isTrash, scopeCounts.global, scopeCounts.project, scopeFilter]);
useEffect(() => {
if (isTrash) return;
if (!isTrash) return;
if (statusFilter === "current" && statusCounts.current === 0) setStatusFilter("all");
if (statusFilter === "open" && statusCounts.open === 0) setStatusFilter("all");
}, [isTrash, statusCounts.current, statusCounts.open, statusFilter]);
useEffect(() => {
if (!isTrash) return;
if (dateFilter !== "all" && dateCounts[dateFilter] === 0) setDateFilter("all");
}, [dateCounts, dateFilter]);
}, [dateCounts, dateFilter, isTrash]);
const filteredSessions = useMemo(() => {
const q = query.trim().toLowerCase();
const q = isTrash ? query.trim().toLowerCase() : "";
return sessions.filter((s) => {
if (scopeFilter !== "all" && sessionScope(s) !== scopeFilter) return false;
if (!isTrash && statusFilter === "current" && !s.current) return false;
@@ -483,6 +501,11 @@ export function HistoryPanel({
onChange={(next) => setDateFilter(next as HistoryDateFilter)}
/>
</div>
{!isTrash && catalogPartial && (
<div className="management-modal__summary history-modal__summary" role="status">
History index is still building ({catalogProgress.indexed}/{catalogProgress.total}); results may be incomplete.
</div>
)}
<div className="history-content">
<div className={`history-list${isTrash ? " history-list--trash" : ""}`}>
@@ -494,7 +517,28 @@ export function HistoryPanel({
) : filteredSessions.length === 0 ? (
<div className="mem-empty">{tr("history.noResults")}</div>
) : (
groups.map((g) => (
<>
{!isTrash && searchHits.length > 0 && (
<section className="mem-section history-search-results">
<div className="mem-section__title hist-group__title">
<span>Content matches</span>
<span className="hist-group__count">{searchHits.length}</span>
</div>
{searchHits.map((hit) => (
<div className="hist-item" key={`${hit.sessionPath}:${hit.messageIndex}:${hit.kind}:${hit.toolName ?? ""}`}>
<button className="hist-item__main" type="button" onClick={() => void loadSearchContext(hit)}>
<div className="hist-item__preview">{hit.sessionTitle || hit.topicTitle || hit.sessionPath}</div>
<div className="hist-item__meta">
<span className="hist-item__badge">{hit.role} · {hit.kind}</span>
{hit.toolName && <span className="hist-item__scope">{hit.toolName}</span>}
</div>
<div className="hist-item__meta">{hit.snippet}</div>
</button>
</div>
))}
</section>
)}
{groups.map((g) => (
<section className="mem-section" key={`${g.recoveryCopy ? "recovery-copy" : "normal"}-${g.label}`}>
<div className="mem-section__title hist-group__title">
<span>{g.recoveryCopy ? `${tr("history.recoveryCopiesGroup")} · ${g.label}` : g.label}</span>
@@ -563,12 +607,45 @@ export function HistoryPanel({
);
})}
</section>
))
))}
{!isTrash && nextCursor && (
<button
className="btn btn--small"
type="button"
onClick={loadMore}
>
Load more
</button>
)}
</>
)}
</div>
<section className={`history-preview${!preview ? " history-preview--empty" : ""}`}>
{preview ? (
<section className={`history-preview${!preview && !searchContext ? " history-preview--empty" : ""}`}>
{searchContext ? (
<>
<div className="history-preview__head">
<div className="history-preview__copy">
<div className="history-preview__title">{searchContext.hit.sessionTitle || searchContext.hit.topicTitle || searchContext.hit.sessionPath}</div>
<div className="history-preview__meta">{searchContext.hit.role} · {searchContext.hit.kind}</div>
</div>
</div>
<div className="history-preview__body">
{searchContext.loading ? (
<div className="mem-empty">{tr("common.loading")}</div>
) : searchContext.lines.length === 0 ? (
<div className="mem-empty">{tr("history.previewEmpty")}</div>
) : (
searchContext.lines.map((line) => (
<div className="hist-item" key={line.index}>
<div className="hist-item__meta"><span className="hist-item__badge">{line.role}</span></div>
<div className="hist-item__preview">{line.text}</div>
</div>
))
)}
</div>
</>
) : preview ? (
<>
<div className="history-preview__head">
<div className="history-preview__copy">
@@ -701,34 +778,3 @@ function sessionMetaLine(s: SessionMeta, tr: ReturnType<typeof useT>, isTrash =
function previewMessagesToItems(messages: HistoryMessage[]): Item[] {
return historyMessagesToItems(messages, "hp").items;
}
function HistoryFilterSelect({
label,
options,
value,
onChange,
}: {
label: string;
options: { id: string; label: string; count: number }[];
value: string;
onChange: (next: string) => void;
}) {
const visibleOptions = options.filter((option) => option.id === "all" || option.id === value || option.count > 0);
return (
<div className="history-filter" role="group" aria-label={label}>
{visibleOptions.map((option) => (
<button
key={option.id}
type="button"
className={`history-filter__pill${value === option.id ? " history-filter__pill--on" : ""}`}
aria-pressed={value === option.id}
disabled={option.id !== "all" && option.count === 0}
onClick={() => onChange(option.id)}
>
{option.label}
<span className="history-filter__count">{option.count}</span>
</button>
))}
</div>
);
}
+3 -1
View File
@@ -8,6 +8,7 @@ import type { InvocationRequest } from "./invocationDisplay";
import { addBreadcrumb } from "./breadcrumbs";
import { maybeShare } from "./queryCoalesce";
import { makeMockSessionCatalogBindings } from "./sessionCatalogBridge";
import { makeMockHistoryCatalogBindings, type HistoryCatalogBindings } from "./historyCatalogBridge";
import { t } from "./i18n";
import { providerIsConfigured, providerRequiresKey, removeProviderAccessesForMock } from "./providerModels";
import { DEFAULT_STATUS_BAR_ITEMS, normalizeStatusBarItems } from "./statusBarItems";
@@ -162,7 +163,7 @@ interface DesktopWindowState {
// added or renamed, the generated types shift, and a key present in GeneratedApp
// but missing from AppBindings causes a type error here. Fix: add the new method
// to AppBindings, then run `pnpm typecheck` to verify.
export interface AppBindings extends SessionCatalogBindings {
export interface AppBindings extends SessionCatalogBindings, HistoryCatalogBindings {
Platform(): Promise<string>;
MinimiseMainWindow(): Promise<void>;
ToggleMaximiseMainWindow(): Promise<void>;
@@ -3262,6 +3263,7 @@ function makeMockApp(): AppBindings {
async ListSessionsForTab() {
return sessions.map((s) => ({ ...s }));
},
...makeMockHistoryCatalogBindings(sessions),
async ListTrashedSessions() {
return trashedSessions.map((s) => ({ ...s }));
},
@@ -0,0 +1,35 @@
import type {
HistoryIndexStatus, HistorySearchContextLine, HistorySearchContextRequest, HistorySearchPage,
HistorySearchRequest, HistorySessionPage, HistorySessionPageRequest,
} from "./historyCatalogTypes";
import type { SessionMeta } from "./types";
export interface HistoryCatalogBindings {
ListHistorySessions(req: HistorySessionPageRequest): Promise<HistorySessionPage>;
SearchHistoryContent(req: HistorySearchRequest): Promise<HistorySearchPage>;
GetHistorySearchContext(req: HistorySearchContextRequest): Promise<HistorySearchContextLine[]>;
GetHistoryIndexStatus(): Promise<HistoryIndexStatus>;
RebuildHistoryIndex(): Promise<void>;
}
export function makeMockHistoryCatalogBindings(sessions: SessionMeta[]): HistoryCatalogBindings {
const status = async (): Promise<HistoryIndexStatus> => ({
state: "ready", mode: "memory", revision: 1, indexed: sessions.length, total: sessions.length, pending: 0, failed: 0,
});
return {
async ListHistorySessions(req) {
const start = req.cursor ? Number(req.cursor) || 0 : 0;
const query = req.query.trim().toLowerCase();
const items = sessions.filter((session) =>
(req.scope === "all" || (session.scope || "global") === req.scope) &&
(req.status !== "current" || session.current) && (req.status !== "open" || session.open) &&
(!query || [session.title, session.preview, session.topicTitle, session.workspaceRoot].some((value) => (value || "").toLowerCase().includes(query))));
const limit = Math.max(1, Math.min(req.limit || 50, 200));
return { items: items.slice(start, start + limit).map((session) => ({ ...session })), nextCursor: start + limit < items.length ? String(start + limit) : "", revision: 1, partial: false, staleCursor: false };
},
async SearchHistoryContent() { return { items: [], nextCursor: "", revision: 1, partial: false, staleCursor: false, status: await status() }; },
async GetHistorySearchContext() { return []; },
GetHistoryIndexStatus: status,
async RebuildHistoryIndex() {},
};
}
@@ -0,0 +1,30 @@
import type { SessionMeta } from "./types";
export interface HistorySessionPageRequest {
scope: "all" | "project" | "global" | string;
workspaceRoot: string;
status: "all" | "current" | "open" | string;
timeFilter: "all" | "today" | "yesterday" | "older" | string;
query: string;
cursor: string;
limit: number;
}
export interface HistorySessionPage { items: SessionMeta[]; nextCursor: string; revision: number; partial: boolean; staleCursor: boolean }
export interface HistoryIndexStatus {
state: string; mode: "disk" | "memory" | string; path?: string; revision: number;
indexed: number; total: number; pending: number; failed: number; lastError?: string; quarantinedPath?: string;
}
export interface HistorySearchRequest {
query: string; scope: "all" | "project" | "global" | string; workspaceRoot: string;
status: "all" | "current" | "open" | string; timeFilter: string; kinds: string[]; toolName: string; cursor: string; limit: number;
}
export interface HistorySearchHit {
sessionPath: string; sessionId: string; source: string; messageIndex: number; role: string; kind: string;
toolName?: string; snippet: string; score: number; sessionTitle?: string; topicTitle?: string; workspaceRoot?: string;
lastActivityAt: number; open: boolean; running: boolean; current: boolean;
}
export interface HistorySearchPage {
items: HistorySearchHit[]; nextCursor: string; revision: number; partial: boolean; staleCursor: boolean; status: HistoryIndexStatus;
}
export interface HistorySearchContextRequest { sessionPath: string; messageIndex: number; before: number; after: number }
export interface HistorySearchContextLine { index: number; role: string; text: string }
+2
View File
@@ -799,6 +799,8 @@ export interface SessionMeta {
recoveryCopy?: boolean; // actual branch content is unchanged and covered by its parent
}
export type { HistoryIndexStatus, HistorySearchContextLine, HistorySearchContextRequest, HistorySearchHit, HistorySearchPage, HistorySearchRequest, HistorySessionPage, HistorySessionPageRequest } from "./historyCatalogTypes";
export interface WorkspaceView {
path: string;
name: string;
+1 -1
View File
@@ -4041,7 +4041,7 @@ export function useController() {
}
}, [activeTabId, bumpCheckpointRefreshSeq, bumpSessionLoadSeq, dispatchTo, loadSessionDataForTab, waitForTabReady]);
const listSessions = useCallback(async (): Promise<SessionMeta[]> => asArray<SessionMeta>(await app.ListSessions().catch(() => [])), []);
const listSessions = useCallback(async (): Promise<SessionMeta[]> => asArray<SessionMeta>((await app.ListHistorySessions({ scope: "all", workspaceRoot: "", status: "all", timeFilter: "all", query: "", cursor: "", limit: 200 }).catch(() => null))?.items), []);
const listTrashedSessions = useCallback(async (): Promise<SessionMeta[]> => asArray<SessionMeta>(await app.ListTrashedSessions().catch(() => [])), []);
const resumeSession = useCallback(async (path: string, tabId?: string, navigationIntentSeq?: number) => {
const targetTabId = tabId || activeTabId;
@@ -0,0 +1,99 @@
import { useCallback, useEffect, useRef, useState } from "react";
import { app } from "./bridge";
import type { HistorySearchHit, SessionMeta } from "./types";
export function useHistoryCatalog({
isTrash,
suppliedSessions,
scope,
status,
timeFilter,
query,
}: {
isTrash: boolean;
suppliedSessions: SessionMeta[];
scope: string;
status: string;
timeFilter: string;
query: string;
}) {
const [catalogSessions, setCatalogSessions] = useState<SessionMeta[]>(suppliedSessions);
const [nextCursor, setNextCursor] = useState("");
const [partial, setPartial] = useState(false);
const [progress, setProgress] = useState({ indexed: 0, total: 0 });
const [searchHits, setSearchHits] = useState<HistorySearchHit[]>([]);
const [refreshNonce, setRefreshNonce] = useState(0);
const requestSeq = useRef(0);
const revision = useRef(0);
const refreshTimer = useRef<number | null>(null);
const workspaceRoot = suppliedSessions.find((session) => session.current)?.workspaceRoot ?? "";
useEffect(() => {
if (isTrash) setCatalogSessions(suppliedSessions);
}, [isTrash, suppliedSessions]);
const fetchPage = useCallback(async (cursor: string, append: boolean, seq: number) => {
const request = { scope, workspaceRoot, status, timeFilter, query: query.trim(), cursor, limit: 50 };
const [page, bodyPage] = await Promise.all([
app.ListHistorySessions(request),
query.trim() ? app.SearchHistoryContent({ ...request, kinds: [], toolName: "" }) : Promise.resolve(null),
]);
if (seq !== requestSeq.current) return;
if (page.staleCursor && cursor) {
void fetchPage("", false, seq);
return;
}
setCatalogSessions((current) => append
? [...current, ...page.items.filter((item) => !current.some((existing) => existing.path === item.path))]
: page.items);
setNextCursor(page.nextCursor);
setPartial(page.partial || Boolean(bodyPage?.partial));
if (!append) setSearchHits(bodyPage?.items ?? []);
if (bodyPage) setProgress({ indexed: bodyPage.status.indexed, total: bodyPage.status.total });
}, [query, scope, status, timeFilter, workspaceRoot]);
useEffect(() => {
if (isTrash || typeof window === "undefined" || !window.runtime) return;
const seq = ++requestSeq.current;
const timer = window.setTimeout(() => {
void fetchPage("", false, seq).catch(() => {
if (seq !== requestSeq.current) return;
setCatalogSessions(suppliedSessions);
setNextCursor("");
setSearchHits([]);
});
}, query.trim() ? 200 : 0);
return () => window.clearTimeout(timer);
}, [fetchPage, isTrash, query, refreshNonce, suppliedSessions]);
useEffect(() => {
if (isTrash || typeof window === "undefined" || !window.runtime) return;
const unsubscribe = window.runtime.EventsOn("history-index:changed-v1", (payload?: unknown) => {
if (!payload || typeof payload !== "object") return;
const event = payload as { revision?: number; indexed?: number; total?: number; pending?: number };
const nextRevision = typeof event.revision === "number" ? event.revision : 0;
if (nextRevision <= revision.current) return;
revision.current = nextRevision;
setProgress({ indexed: Number(event.indexed) || 0, total: Number(event.total) || 0 });
setPartial((Number(event.pending) || 0) > 0 || (Number(event.indexed) || 0) < (Number(event.total) || 0));
if (refreshTimer.current === null) {
refreshTimer.current = window.setTimeout(() => {
refreshTimer.current = null;
setRefreshNonce((value) => value + 1);
}, 200);
}
});
return () => {
unsubscribe();
if (refreshTimer.current !== null) window.clearTimeout(refreshTimer.current);
refreshTimer.current = null;
};
}, [isTrash]);
const loadMore = useCallback(() => {
if (!nextCursor) return;
void fetchPage(nextCursor, true, requestSeq.current);
}, [fetchPage, nextCursor]);
return { sessions: isTrash ? suppliedSessions : catalogSessions, nextCursor, partial, progress, searchHits, loadMore };
}
+440
View File
@@ -0,0 +1,440 @@
package main
import (
"context"
"encoding/base64"
"encoding/json"
"errors"
"path/filepath"
"strings"
"time"
"reasonix/internal/agent"
"reasonix/internal/config"
"reasonix/internal/history"
"reasonix/internal/historycatalog"
"reasonix/internal/provider"
"reasonix/internal/retrieval"
"reasonix/internal/sessioncatalog"
)
type HistorySessionPageRequest struct {
Scope string `json:"scope"`
WorkspaceRoot string `json:"workspaceRoot,omitempty"`
Status string `json:"status"`
TimeFilter string `json:"timeFilter"`
Query string `json:"query"`
Cursor string `json:"cursor"`
Limit int `json:"limit"`
}
type HistorySessionPage struct {
Items []SessionMeta `json:"items"`
NextCursor string `json:"nextCursor"`
Revision uint64 `json:"revision"`
Partial bool `json:"partial"`
StaleCursor bool `json:"staleCursor"`
}
type HistorySearchRequest struct {
Query string `json:"query"`
Scope string `json:"scope"`
WorkspaceRoot string `json:"workspaceRoot,omitempty"`
Status string `json:"status"`
TimeFilter string `json:"timeFilter"`
Kinds []string `json:"kinds"`
ToolName string `json:"toolName"`
Cursor string `json:"cursor"`
Limit int `json:"limit"`
}
type HistorySearchHit struct {
SessionPath string `json:"sessionPath"`
SessionID string `json:"sessionId"`
Source string `json:"source"`
MessageIndex int `json:"messageIndex"`
Role string `json:"role"`
Kind string `json:"kind"`
ToolName string `json:"toolName,omitempty"`
Snippet string `json:"snippet"`
Score float64 `json:"score"`
SessionTitle string `json:"sessionTitle,omitempty"`
TopicTitle string `json:"topicTitle,omitempty"`
WorkspaceRoot string `json:"workspaceRoot,omitempty"`
LastActivityAt int64 `json:"lastActivityAt"`
Open bool `json:"open"`
Running bool `json:"running"`
Current bool `json:"current"`
}
type HistoryIndexStatus = historycatalog.Status
type HistoryIndexChangedV1 struct {
Revision uint64 `json:"revision"`
Indexed int64 `json:"indexed"`
Total int64 `json:"total"`
Pending int64 `json:"pending"`
Roots []string `json:"roots"`
Reason string `json:"reason"`
}
func (a *App) registerHistoryIndexEvents() {
history.RegisterCatalogObserver(func(status historycatalog.Status, roots []string, reason string) {
if roots == nil {
roots = []string{}
}
a.emitRuntimeEvent("history-index:changed-v1", HistoryIndexChangedV1{Revision: status.Revision,
Indexed: status.Indexed, Total: status.Total, Pending: status.Pending, Roots: roots, Reason: reason})
})
}
type HistorySearchPage struct {
Items []HistorySearchHit `json:"items"`
NextCursor string `json:"nextCursor"`
Revision uint64 `json:"revision"`
Partial bool `json:"partial"`
StaleCursor bool `json:"staleCursor"`
Status HistoryIndexStatus `json:"status"`
}
type HistorySearchContextRequest struct {
SessionPath string `json:"sessionPath"`
MessageIndex int `json:"messageIndex"`
Before int `json:"before"`
After int `json:"after"`
}
type HistorySearchContextLine struct {
Index int `json:"index"`
Role string `json:"role"`
Text string `json:"text"`
}
type historySearchCursor struct {
Revision uint64 `json:"r"`
Rank float64 `json:"b"`
Score float64 `json:"s"`
Path string `json:"p"`
Message int `json:"m"`
Part int `json:"a"`
RowID int64 `json:"i"`
}
func historyCatalogRoots(targets []sessioncatalog.DirectoryTarget) []historycatalog.Root {
roots := make([]historycatalog.Root, 0, len(targets)*2+1)
for _, target := range targets {
source := target.Scope
if source == "" {
source = "global"
}
root := historycatalog.Root{Path: target.Path, Source: source, Scope: target.Scope, WorkspaceRoot: target.WorkspaceRoot}
roots = append(roots, root)
root.Path = filepath.Join(target.Path, "subagents")
root.Subagents = true
roots = append(roots, root)
}
roots = append(roots, historycatalog.Root{Path: config.ArchiveDir(), Source: "archive", Scope: "global", Archive: true})
return roots
}
func sessionMetaFromCatalog(record sessioncatalog.SessionRecord, current, open bool) SessionMeta {
title := strings.TrimSpace(record.CustomTitle)
preview := record.Preview
if strings.TrimSpace(preview) == "" && record.TurnsState == sessioncatalog.TurnsUnknown {
preview = "History is being indexed — " + filepath.Base(record.Path)
}
recovered := record.Recovered || strings.TrimSpace(record.RecoveryDigest) != "" || isAutomaticRecoverySessionPath(record.Path)
return SessionMeta{Path: record.Path, Preview: preview, Title: title, Turns: record.Turns,
TurnsState: string(record.TurnsState), CreatedAt: record.CreatedAt, LastActivityAt: record.LastActivityAt,
ModTime: record.LastActivityAt, Current: current, Open: open, Scope: record.Scope,
WorkspaceRoot: record.WorkspaceRoot, TopicID: record.TopicID, TopicTitle: record.TopicTitle,
Recovered: recovered}
}
func (a *App) ListHistorySessions(req HistorySessionPageRequest) HistorySessionPage {
out := HistorySessionPage{Items: []SessionMeta{}}
catalog := a.sessionCatalog.Load()
if catalog == nil {
out.Partial = true
return out
}
page, err := catalog.ListSessions(a.bootContext(), sessioncatalog.SessionPageRequest{
Scope: req.Scope, WorkspaceRoot: req.WorkspaceRoot, Cursor: req.Cursor, Limit: req.Limit,
Query: req.Query, TimeFilter: req.TimeFilter,
})
if err != nil {
out.Partial = true
return out
}
_, overlays := a.catalogRuntimeOverlays()
active := a.activeSessionPath(a.activeSessionDir())
for _, record := range page.Items {
overlay := overlays[sessionRuntimeKey(record.Path)]
if req.Status == "open" && !overlay.open {
continue
}
if req.Status == "current" && record.Path != active {
continue
}
out.Items = append(out.Items, sessionMetaFromCatalog(record, record.Path == active, overlay.open))
}
out.NextCursor, out.Revision, out.StaleCursor = page.NextCursor, page.Revision, page.StaleCursor
status := catalog.Status()
out.Partial = status.State != sessioncatalog.StateReady || status.Indexed < status.Total
return out
}
func (a *App) GetHistoryIndexStatus() HistoryIndexStatus {
if catalog := history.SharedCatalog(); catalog != nil {
return catalog.Status()
}
return HistoryIndexStatus{State: "opening", Pending: 1, Mode: "memory"}
}
func decodeHistoryCursor(encoded string) (*historySearchCursor, error) {
if strings.TrimSpace(encoded) == "" {
return nil, nil
}
b, err := base64.RawURLEncoding.DecodeString(encoded)
if err != nil {
return nil, err
}
var cursor historySearchCursor
if err := json.Unmarshal(b, &cursor); err != nil || cursor.Path == "" {
return nil, errors.New("invalid history search cursor")
}
return &cursor, nil
}
func encodeHistoryCursor(cursor historySearchCursor) string {
b, _ := json.Marshal(cursor)
return base64.RawURLEncoding.EncodeToString(b)
}
func historyCandidateAfter(candidate historycatalog.Candidate, cursor *historySearchCursor) bool {
if cursor == nil {
return true
}
if candidate.Rank != cursor.Rank {
return candidate.Rank > cursor.Rank
}
if candidate.SessionPath != cursor.Path {
return candidate.SessionPath > cursor.Path
}
if candidate.MessageIndex != cursor.Message {
return candidate.MessageIndex > cursor.Message
}
if candidate.PartIndex != cursor.Part {
return candidate.PartIndex > cursor.Part
}
return candidate.RowID > cursor.RowID
}
func desktopHistoryText(messages []provider.Message, candidate historycatalog.Candidate) (string, bool) {
if candidate.MessageIndex < 0 || candidate.MessageIndex >= len(messages) {
return "", false
}
message := messages[candidate.MessageIndex]
switch candidate.Kind {
case "user_text", "assistant_text":
return message.Content, true
case "tool_input":
if candidate.PartIndex < 0 || candidate.PartIndex >= len(message.ToolCalls) {
return "", false
}
call := message.ToolCalls[candidate.PartIndex]
return strings.TrimSpace(call.Name + " " + call.Arguments), true
case "tool_error", "tool_output":
return strings.TrimSpace(message.Name + " " + message.Content), true
default:
return "", false
}
}
func (a *App) SearchHistoryContent(req HistorySearchRequest) HistorySearchPage {
status := a.GetHistoryIndexStatus()
out := HistorySearchPage{Items: []HistorySearchHit{}, Status: status, Revision: status.Revision,
Partial: status.State != "ready" || status.Pending > 0 || (status.Total > 0 && status.Indexed < status.Total)}
catalog := history.SharedCatalog()
if catalog == nil || strings.TrimSpace(req.Query) == "" {
return out
}
cursor, err := decodeHistoryCursor(req.Cursor)
if err != nil {
return out
}
if cursor != nil && cursor.Revision != status.Revision {
out.StaleCursor = true
return out
}
limit := req.Limit
if limit <= 0 {
limit = 50
}
if limit > 200 {
limit = 200
}
kinds := req.Kinds
if len(kinds) == 0 {
kinds = []string{"user_text", "assistant_text", "tool_input", "tool_error"}
}
rootFilter := []string{}
for _, root := range historyCatalogRoots(a.sessionCatalogTargets()) {
if req.Scope == "project" && (root.Scope != "project" || !sameProjectRoot(root.WorkspaceRoot, req.WorkspaceRoot)) {
continue
}
if req.Scope == "global" && root.Scope == "project" {
continue
}
rootFilter = append(rootFilter, root.Path)
}
var after *historycatalog.SearchCursor
if cursor != nil {
after = &historycatalog.SearchCursor{Rank: cursor.Rank, SessionPath: cursor.Path, MessageIndex: cursor.Message,
PartIndex: cursor.Part, RowID: cursor.RowID}
}
result, err := catalog.Search(a.bootContext(), historycatalog.SearchRequest{Query: req.Query, Scope: req.Scope,
WorkspaceRoot: req.WorkspaceRoot, Kinds: kinds, ToolName: req.ToolName, Limit: 200, Roots: rootFilter, After: after})
if err != nil {
out.Status.LastError = err.Error()
return out
}
out.Items, out.NextCursor = a.collectHistoryHits(req, result.Items, cursor, limit, status.Revision, catalog)
return out
}
func (a *App) collectHistoryHits(req HistorySearchRequest, candidates []historycatalog.Candidate, cursor *historySearchCursor,
limit int, revision uint64, catalog *historycatalog.Catalog) ([]HistorySearchHit, string) {
_, overlays := a.catalogRuntimeOverlays()
active := a.activeSessionPath(a.activeSessionDir())
queryTerms, _ := retrieval.QueryTerms(req.Query)
loaded := map[string][]provider.Message{}
items := []HistorySearchHit{}
var cursorCandidate historycatalog.Candidate
for _, candidate := range candidates {
if !historyCandidateAfter(candidate, cursor) {
continue
}
overlay := overlays[sessionRuntimeKey(candidate.SessionPath)]
if req.Status == "open" && !overlay.open || req.Status == "current" && candidate.SessionPath != active {
continue
}
if !historyTimeMatches(candidate.LastActivityAt, req.TimeFilter) {
continue
}
messages, ok := loaded[candidate.SessionPath]
if !ok {
if identity, known, identityErr := agent.SessionContentIdentity(candidate.SessionPath); identityErr == nil && known &&
candidate.ContentDigest != "" && identity.DigestHex != candidate.ContentDigest {
catalog.EnqueueExisting(context.Background(), candidate.SessionPath)
continue
}
session, loadErr := agent.LoadSession(candidate.SessionPath)
if loadErr != nil {
continue
}
messages = session.Snapshot()
loaded[candidate.SessionPath] = messages
}
text, ok := desktopHistoryText(messages, candidate)
if !ok {
catalog.EnqueueExisting(context.Background(), candidate.SessionPath)
continue
}
items = append(items, HistorySearchHit{SessionPath: candidate.SessionPath,
SessionID: strings.TrimSuffix(filepath.Base(candidate.SessionPath), filepath.Ext(candidate.SessionPath)), Source: candidate.Source,
MessageIndex: candidate.MessageIndex, Role: candidate.Role, Kind: candidate.Kind, ToolName: candidate.ToolName,
Snippet: retrieval.MakeSnippet(text, req.Query, queryTerms, 240), Score: candidate.Score,
SessionTitle: candidate.SessionTitle, TopicTitle: candidate.TopicTitle, WorkspaceRoot: candidate.WorkspaceRoot,
LastActivityAt: candidate.LastActivityAt, Open: overlay.open, Running: overlay.running, Current: candidate.SessionPath == active})
if len(items) == limit {
cursorCandidate = candidate
}
if len(items) == limit+1 {
break
}
}
nextCursor := ""
if len(items) > limit {
items = items[:limit]
nextCursor = encodeHistoryCursor(historySearchCursor{Revision: revision, Rank: cursorCandidate.Rank, Score: cursorCandidate.Score, Path: cursorCandidate.SessionPath,
Message: cursorCandidate.MessageIndex, Part: cursorCandidate.PartIndex, RowID: cursorCandidate.RowID})
}
return items, nextCursor
}
func historyTimeMatches(timestamp int64, filter string) bool {
if strings.TrimSpace(filter) == "" || filter == "all" {
return true
}
now := time.Now()
startToday := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location())
value := time.UnixMilli(timestamp)
switch filter {
case "today":
return !value.Before(startToday)
case "yesterday":
return !value.Before(startToday.AddDate(0, 0, -1)) && value.Before(startToday)
case "older":
return value.Before(startToday.AddDate(0, 0, -1))
default:
return true
}
}
func (a *App) GetHistorySearchContext(req HistorySearchContextRequest) []HistorySearchContextLine {
var options history.Options
bestLength := -1
for _, root := range historyCatalogRoots(a.sessionCatalogTargets()) {
if !desktopHistoryPathWithin(req.SessionPath, root.Path) || len(root.Path) <= bestLength {
continue
}
bestLength = len(root.Path)
options = history.Options{}
switch {
case root.Archive:
options.ArchiveDir = root.Path
case root.Subagents:
options.SessionDir = filepath.Dir(root.Path)
default:
options.SessionDir = root.Path
}
}
if bestLength < 0 {
return []HistorySearchContextLine{}
}
searcher := history.NewSearcher(options)
lines, err := searcher.Around(a.bootContext(), history.AroundRequest{SessionPath: req.SessionPath, MessageIndex: req.MessageIndex, Before: req.Before, After: req.After})
if err != nil {
return []HistorySearchContextLine{}
}
out := make([]HistorySearchContextLine, 0, len(lines))
for _, line := range lines {
role := ""
if fields := strings.Fields(line.Text); len(fields) > 1 {
role = strings.TrimSuffix(fields[1], "]")
}
out = append(out, HistorySearchContextLine{Index: line.Index, Role: role, Text: line.Text})
}
return out
}
func desktopHistoryPathWithin(path, root string) bool {
absPath, err := filepath.Abs(filepath.Clean(strings.TrimSpace(path)))
if err != nil {
return false
}
absRoot, err := filepath.Abs(filepath.Clean(strings.TrimSpace(root)))
if err != nil {
return false
}
rel, err := filepath.Rel(absRoot, absPath)
return err == nil && (rel == "." || (rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator))))
}
func (a *App) RebuildHistoryIndex() error {
for _, root := range historyCatalogRoots(a.sessionCatalogTargets()) {
history.RegisterCatalogRoots([]historycatalog.Root{root})
}
return nil
}
+39 -1
View File
@@ -8,7 +8,9 @@ import (
"strings"
"time"
"reasonix/internal/agent"
"reasonix/internal/config"
"reasonix/internal/history"
"reasonix/internal/sessioncatalog"
)
@@ -59,6 +61,10 @@ type ProjectTreeChangedV2 struct {
Reason string `json:"reason"`
}
func flushDesktopDerivedCatalogs(ctx context.Context) error {
return history.FlushSharedCatalog(ctx)
}
func sessionCatalogStatus(status sessioncatalog.Status) SessionCatalogStatus {
return SessionCatalogStatus{
State: string(status.State),
@@ -106,6 +112,7 @@ func (a *App) startSessionCatalog(rebuild bool) {
defer a.catalogRebuilding.Store(false)
path := sessioncatalog.DefaultPath()
targets := a.sessionCatalogTargets()
history.RegisterCatalogRoots(historyCatalogRoots(targets))
if rebuild {
if _, err := sessioncatalog.Rebuild(ctx, path, targets); err != nil && !errors.Is(err, context.Canceled) {
slog.Warn("desktop: rebuild session catalog", "err", err)
@@ -240,6 +247,30 @@ func (a *App) sessionCatalogTargets() []sessioncatalog.DirectoryTarget {
return out
}
func listCatalogSessionsForDirectory(ctx context.Context, catalog *sessioncatalog.Catalog,
target sessioncatalog.DirectoryTarget, directory string) ([]sessioncatalog.SessionRecord, error) {
for range 2 {
records := []sessioncatalog.SessionRecord{}
cursor := ""
for {
page, err := catalog.ListSessions(ctx, sessioncatalog.SessionPageRequest{Scope: target.Scope,
WorkspaceRoot: target.WorkspaceRoot, Directory: directory, Cursor: cursor, Limit: sessioncatalog.MaxLimit})
if err != nil {
return nil, err
}
if page.StaleCursor {
break
}
records = append(records, page.Items...)
if page.NextCursor == "" {
return records, nil
}
cursor = page.NextCursor
}
}
return []sessioncatalog.SessionRecord{}, nil
}
func (a *App) indexRestoredSessionPaths(ctx context.Context, catalog *sessioncatalog.Catalog) {
type restored struct {
target sessioncatalog.DirectoryTarget
@@ -361,6 +392,9 @@ func sessionDirectoryForPath(path string) string {
}
func (a *App) requestSessionCatalogPath(scope, workspaceRoot, path string) {
if strings.TrimSpace(path) != "" {
_ = history.PersistObserver().EnqueueSessionPersist(agent.SessionPersistEvent{Path: path, Rewrite: true})
}
catalog := a.sessionCatalog.Load()
if catalog == nil || a.shuttingDown.Load() || strings.TrimSpace(path) == "" {
return
@@ -371,8 +405,12 @@ func (a *App) requestSessionCatalogPath(scope, workspaceRoot, path string) {
}
func (a *App) removeSessionCatalogPath(path, reason string) {
if strings.TrimSpace(path) == "" {
return
}
_ = history.PersistObserver().EnqueueSessionPersist(agent.SessionPersistEvent{Path: path, Removed: true})
catalog := a.sessionCatalog.Load()
if catalog == nil || strings.TrimSpace(path) == "" {
if catalog == nil {
return
}
ctx, cancel := context.WithTimeout(a.bootContext(), 150*time.Millisecond)
+36
View File
@@ -1,11 +1,47 @@
package main
import (
"context"
"os"
"path/filepath"
"testing"
"time"
"reasonix/internal/sessioncatalog"
)
func installSessionCatalogForTest(t *testing.T, app *App, path, scope, workspaceRoot string) {
t.Helper()
catalog, err := sessioncatalog.Open(context.Background(), sessioncatalog.Options{InMemory: true, DisableRepair: true})
if err != nil {
t.Fatalf("open in-memory session catalog: %v", err)
}
target := sessioncatalog.DirectoryTarget{Path: path, Scope: scope, WorkspaceRoot: workspaceRoot}
if err := catalog.ReconcileDirectory(context.Background(), target); err != nil {
_ = catalog.Close(context.Background())
t.Fatalf("reconcile session catalog %q: %v", target.Path, err)
}
app.sessionCatalog.Store(catalog)
t.Cleanup(func() {
app.sessionCatalog.CompareAndSwap(catalog, nil)
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
_ = catalog.Close(ctx)
})
}
func reconcileSessionCatalogForTest(t *testing.T, app *App, path, scope, workspaceRoot string) {
t.Helper()
catalog := app.sessionCatalog.Load()
if catalog == nil {
t.Fatal("session catalog is not installed")
}
target := sessioncatalog.DirectoryTarget{Path: path, Scope: scope, WorkspaceRoot: workspaceRoot}
if err := catalog.ReconcileDirectory(context.Background(), target); err != nil {
t.Fatalf("reconcile session catalog %q: %v", path, err)
}
}
func TestProjectTreeSnapshotReturnsProjectShellWithoutMigratingSessions(t *testing.T) {
isolateDesktopUserDirs(t)
root := t.TempDir()
+1 -1
View File
@@ -852,7 +852,7 @@ func TestHistoryMarksLegacyRecoverySessionsAsRecovered(t *testing.T) {
defer ctrl.Close()
app := NewApp()
app.setTestCtrl(ctrl, "")
installSessionCatalogForTest(t, app, dir, "global", "")
sessions := app.ListSessions()
for _, session := range sessions {
if filepath.Clean(session.Path) != filepath.Clean(recovery) {
+1
View File
@@ -225,6 +225,7 @@ func TestListSessionsForTabKeepsSourceDirectoryAfterActiveTabSwitch(t *testing.T
},
activeTabID: "tab-b",
}
installSessionCatalogForTest(t, app, dirA, "global", "")
sessions := app.ListSessionsForTab("tab-a")
if len(sessions) != 1 || sessions[0].Path != pathA || sessions[0].TurnsState != "unknown" {
+9
View File
@@ -468,6 +468,15 @@ Without `--dir`, reindex includes global sessions and all projects saved by the
desktop app. See [Session Catalog and Desktop Startup](./SESSION_CATALOG.md) for
failure, migration, and data-safety guarantees.
History search uses a separate disposable projection:
```sh
reasonix doctor catalogs [--json]
reasonix catalogs reindex history [--dir PATH ...] [--json]
```
See [History Search Catalog](./HISTORY_SEARCH_CATALOG.md).
### Memory diagnostics and recovery
Bare `/memory` shows all active project/global facts without hiding same-name
+9
View File
@@ -406,6 +406,15 @@ reasonix sessions reindex --dir /path/to/sessions --dir /another/path
不传 `--dir` 时,reindex 会覆盖全局会话和桌面保存的全部项目。迁移、失败降级和数据安全
契约见 [Session Catalog and Desktop Startup](./SESSION_CATALOG.md)。
历史搜索使用独立的可丢弃投影:
```sh
reasonix doctor catalogs [--json]
reasonix catalogs reindex history [--dir PATH ...] [--json]
```
详见 [历史搜索 Catalog](./HISTORY_SEARCH_CATALOG.zh-CN.md)。
### 记忆诊断与恢复
直接运行 `/memory` 会显示全部 project/global active facts,不会隐藏跨 scope 的同名条目。
+3
View File
@@ -41,6 +41,7 @@ non-destructively when `<Reasonix home>/.env` is missing them.
| Archives | `<state root>/archive/` |
| Memory | `<state root>/memory/` and `<state root>/projects/` |
| Disposable session catalog | `<cache root>/session-catalog/v1.sqlite` |
| Disposable history search catalog | `<cache root>/history-search/v1.sqlite` |
`<state root>` defaults to `<Reasonix home>`. It only differs when
`REASONIX_STATE_HOME` is set.
@@ -48,6 +49,8 @@ non-destructively when `<Reasonix home>/.env` is missing them.
The session catalog is a rebuildable query projection, not user data. Session
JSONL, event logs, metadata sidecars, and `desktop-projects.json` remain
authoritative. See [Session Catalog and Desktop Startup](./SESSION_CATALOG.md).
The history projection is documented in
[History Search Catalog](./HISTORY_SEARCH_CATALOG.md).
The global user config is named `config.toml`. Project-local config files keep
the name `reasonix.toml`. If someone says "global reasonix.toml", they usually
+44
View File
@@ -0,0 +1,44 @@
# History Search Catalog
Reasonix stores the history search projection at
`<cache root>/history-search/v1.sqlite`. Session JSONL, event logs, metadata,
sub-agent transcripts, and archives remain authoritative. Removing or
rebuilding the database never removes conversation data, and older Reasonix
versions continue to read the same authoritative files.
The catalog stores normalized retrieval tokens in FTS5, not complete message
text. English tokens retain the existing lowercase/code-symbol semantics and
CJK text retains overlapping bigrams. Snippets and `around` context are loaded
from the authoritative source only after SQLite has selected the final
candidates. User, assistant, tool input, tool error, and tool output parts are
indexed; ordinary tool output remains excluded by the default search kinds.
Successful session persistence sends a non-blocking, path-coalesced hint after
the authoritative commit and after file locks have been released. Continuous
appends read only the new display-index range. Rewrites, missed notifications,
external writers, or fingerprint mismatches rebuild that source in the
background. Only one full transcript decoder runs at once, checkpoints are
persisted per source, and one corrupt source cannot stop other histories.
The Agent `history` tool and Desktop history manager share the projection.
Search never starts a synchronous directory scan. While indexing is incomplete,
existing results return immediately with explicit progress. Runtime open,
running, and current state is overlaid from memory; SQLite cannot restore stale
runtime ownership. Provider-visible tool name, description, schema, defaults,
and ordering are unchanged.
The database uses the common disposable projection policy: local disks use WAL,
`synchronous=NORMAL`, foreign keys, private permissions, and a short busy
timeout. Remote or unavailable cache directories fall back to memory. Integrity
or migration failures quarantine and rebuild the cache; future schema versions
are preserved in degraded read mode.
Diagnostics and safe rebuild commands:
```sh
reasonix doctor catalogs [--json]
reasonix catalogs reindex history [--dir PATH ...] [--json]
```
Diagnostics never print queries, tokens, snippets, messages, tool arguments, or
provider content. A rebuild replaces only this disposable projection.
+34
View File
@@ -0,0 +1,34 @@
# 历史搜索 Catalog
Reasonix 将历史搜索投影保存到 `<cache root>/history-search/v1.sqlite`。
Session JSONL、event log、metadata、子 Agent transcript 和 archive 仍是唯一权威
数据。删除或重建数据库不会删除会话,旧版 Reasonix 也继续读取原有权威文件。
Catalog 的 FTS5 只保存规范化检索 token,不保存完整消息正文。英文沿用现有小写与
代码符号语义,CJK 沿用重叠 bigram。SQLite 选出最终候选后,snippet 和 `around`
上下文才从权威文件读取。user、assistant、tool input、tool error、tool output 都会索引,
但普通 tool output 仍不属于默认搜索种类。
权威会话提交成功且文件锁释放后,保存路径只发送非阻塞、按 path 合并的索引提示。
连续 append 只读取 display index 的新增范围;rewrite、通知丢失、外部进程写入或指纹
不一致会在后台重建单个 source。完整 transcript decoder 始终单并发,checkpoint 按
source 持久化,单个坏会话不会阻断其他历史。
Agent 的 `history` 工具与 Desktop 历史管理器共享该投影。搜索不会同步扫描目录;首次
索引未完成时立即返回已有结果和明确进度。open、running、current 等运行态只从内存
覆盖,SQLite 不能恢复陈旧所有权。面向 provider 的工具名、描述、schema、默认值和
顺序保持逐字节不变。
数据库遵循通用可丢弃投影策略:本地盘使用 WAL、`synchronous=NORMAL`、外键、用户
私有权限和短 busy timeout;远程或不可用缓存自动退化为内存。完整性或迁移失败只隔离
并重建缓存;遇到未来 schema 时保留原库并进入 degraded。
诊断与安全重建命令:
```sh
reasonix doctor catalogs [--json]
reasonix catalogs reindex history [--dir PATH ...] [--json]
```
诊断不会输出 query、token、snippet、消息、tool arguments 或 provider 内容;reindex
只替换该可丢弃投影。
+4 -18
View File
@@ -192,42 +192,28 @@ type RecoveryBranchInfo struct {
// the append-only event log is authoritative once present, while the .jsonl
// checkpoint also serves as the random-read model for history paging.
func (s *Session) Save(path string) error {
return s.save(path, sessionSaveSnapshot)
return s.saveObserved(path, sessionSaveSnapshot)
}
// SaveSnapshot writes a normal autosave/snapshot only when doing so cannot hide
// a newer transcript already on disk. Explicit history rewrites such as rewind,
// compaction, and cancel recovery should call SaveRewrite instead.
func (s *Session) SaveSnapshot(path string) error {
return s.save(path, sessionSaveSnapshot)
return s.saveObserved(path, sessionSaveSnapshot)
}
// SaveRewrite writes an intentional non-append history rewrite only while this
// Session still owns the current on-disk transcript baseline. It prevents a
// stale controller from force-rewinding a newer transcript written elsewhere.
func (s *Session) SaveRewrite(path string) error {
return s.save(path, sessionSaveRewrite)
return s.saveObserved(path, sessionSaveRewrite)
}
// SaveRewriteCompact performs a CAS-protected rewrite and folds the event log
// to one replace record. It is for destructive maintenance such as redaction:
// retaining old WAL records would keep the removed bytes recoverable on disk.
func (s *Session) SaveRewriteCompact(path string) error {
return s.save(path, sessionSaveRewriteCompact)
}
// SaveIfAbsent persists a newly imported or copied session without ever
// replacing a destination another runtime created after the caller's scan.
// The existence check and the full write share the same in-process and
// cross-process save locks, so migration cannot regress a newer destination
// during startup or an overlapping import.
func (s *Session) SaveIfAbsent(path string) error {
return s.withSessionSaveLocks(path, func() error {
if sessionArtifactExists(path) {
return os.ErrExist
}
return s.saveLocked(path, sessionSaveSnapshot)
})
return s.saveObserved(path, sessionSaveRewriteCompact)
}
func (s *Session) save(path string, mode sessionSaveMode) error {
+4
View File
@@ -54,6 +54,10 @@ type Session struct {
// cache-prefix change. DrainContentRewriteReasons (run_loop.go, once per
// provider request) is the sole consumer.
pendingContentReasons []string
// persistObserver receives non-blocking post-commit projection hints. It is
// deliberately session-local so multiple runtimes cannot steal each other's
// observer registration.
persistObserver SessionPersistObserver
}
// NewSession initializes a session with an optional system prompt.
+52
View File
@@ -0,0 +1,52 @@
package agent
import (
"encoding/json"
"fmt"
"io"
"os"
"reasonix/internal/provider"
"reasonix/internal/store"
)
// LoadSessionDisplayMessageRange decodes a bounded range through the display
// index. Callers must separately compare the index revision and digest with
// the authoritative content ledger before trusting it.
func LoadSessionDisplayMessageRange(path string, start, end int) ([]provider.Message, *SessionDisplayIndex, error) {
index, err := LoadSessionDisplayIndex(store.SessionDisplayIndex(path))
if err != nil {
return nil, nil, err
}
if start < 0 || end < start || end > index.MessageCount {
return nil, index, fmt.Errorf("invalid display message range [%d,%d)", start, end)
}
f, err := os.Open(path)
if err != nil {
return nil, index, err
}
defer f.Close()
info, err := f.Stat()
if err != nil {
return nil, index, err
}
if info.Size() != index.TranscriptSize {
return nil, index, fmt.Errorf("display index transcript size changed")
}
out := make([]provider.Message, 0, end-start)
for _, entry := range index.Entries[start:end] {
line := make([]byte, entry.Length)
if _, err := f.ReadAt(line, entry.Offset); err != nil && err != io.EOF {
return nil, index, err
}
if len(line) == 0 || line[len(line)-1] != '\n' {
return nil, index, fmt.Errorf("display message %d is not newline terminated", entry.Index)
}
var message provider.Message
if err := json.Unmarshal(line[:len(line)-1], &message); err != nil {
return nil, index, fmt.Errorf("decode display message %d: %w", entry.Index, err)
}
out = append(out, message)
}
return out, index, nil
}
@@ -0,0 +1,82 @@
package agent
import (
"os"
"reasonix/internal/store"
)
// SessionPersistEvent is emitted only after the authoritative transcript,
// event log, metadata ledger, and derived display read model have completed
// their save boundary. Observers must only enqueue work and return immediately.
type SessionPersistEvent struct {
Path string
Revision int64
ContentDigest string
MessageCount int
AppendFrom int
Rewrite bool
Removed bool
}
type SessionPersistObserver interface {
EnqueueSessionPersist(SessionPersistEvent) bool
}
func (s *Session) SetPersistObserver(observer SessionPersistObserver) {
if s == nil {
return
}
s.mu.Lock()
s.persistObserver = observer
s.mu.Unlock()
}
func (s *Session) notifyPersisted(path string, rewrite bool, appendFrom int) {
if s == nil {
return
}
s.mu.RLock()
observer := s.persistObserver
messageCount := len(s.Messages)
s.mu.RUnlock()
if observer == nil {
return
}
state, known := s.PersistedState(path)
event := SessionPersistEvent{Path: path, MessageCount: messageCount, AppendFrom: appendFrom, Rewrite: rewrite}
if known {
event.Revision = state.Revision
event.ContentDigest = state.DigestHex
}
observer.EnqueueSessionPersist(event)
}
// SaveIfAbsent persists a newly imported or copied session without replacing a
// destination another runtime created after the caller's scan.
func (s *Session) SaveIfAbsent(path string) error {
err := s.withSessionSaveLocks(path, func() error {
if sessionArtifactExists(path) {
return os.ErrExist
}
return s.saveLocked(path, sessionSaveSnapshot)
})
if err == nil {
s.notifyPersisted(path, false, -1)
}
return err
}
func (s *Session) saveObserved(path string, mode sessionSaveMode) error {
appendFrom := -1
if mode == sessionSaveSnapshot {
if index, err := LoadSessionDisplayIndex(store.SessionDisplayIndex(path)); err == nil {
appendFrom = index.MessageCount
}
}
err := s.save(path, mode)
if err == nil {
s.notifyPersisted(path, mode != sessionSaveSnapshot, appendFrom)
}
return err
}
+2 -2
View File
@@ -1151,7 +1151,7 @@ func build(ctx context.Context, opts Options) (*BuildResult, error) {
reg.Add(sessiontool.NewReadSessionTool(sessionDir))
return "enabled list_sessions, read_session."
}
reg.Add(history.NewTool(history.Options{SessionDir: sessionDir, GlobalSessionDir: config.SessionDir(), ArchiveDir: config.ArchiveDir()}))
reg.Add(history.NewIndexedTool(history.Options{SessionDir: sessionDir, GlobalSessionDir: config.SessionDir(), ArchiveDir: config.ArchiveDir()}))
reg.Add(sessiontool.NewListSessionsTool(sessionDir))
reg.Add(sessiontool.NewReadSessionTool(sessionDir))
return "enabled history, list_sessions, read_session."
@@ -1809,7 +1809,7 @@ func build(ctx context.Context, opts Options) (*BuildResult, error) {
return missing
})
execSess := agent.NewSession(sysPrompt)
execSess := newObservedSession(sysPrompt)
executor := agent.New(execProv, reg, execSess, agent.Options{
MaxSteps: maxSteps,
MaxStepsKey: opts.MaxStepsKey,
+12
View File
@@ -0,0 +1,12 @@
package boot
import (
"reasonix/internal/agent"
"reasonix/internal/history"
)
func newObservedSession(systemPrompt string) *agent.Session {
session := agent.NewSession(systemPrompt)
session.SetPersistObserver(history.PersistObserver())
return session
}
+108
View File
@@ -0,0 +1,108 @@
package cli
import (
"context"
"encoding/json"
"flag"
"fmt"
"os"
"time"
"reasonix/internal/projectiondb"
"reasonix/internal/sessioncatalog"
)
type catalogInspection struct {
Name string `json:"name"`
Info projectiondb.Inspection `json:"info"`
}
type catalogCommand struct {
name string
path func() string
reindex func([]string) int
completionFlags []cliCompletionFlag
}
var catalogCommands []catalogCommand
func registerCatalogCommand(command catalogCommand) {
catalogCommands = append(catalogCommands, command)
}
func runSessionOrCatalogCommand(command string, args []string) int {
configureCLIThemeFromConfig()
if command != "catalogs" {
return sessionOrSessionsCommand(command, args)
}
return catalogsCommand(args)
}
func doctorCatalogsCommand(args []string) int {
fs := flag.NewFlagSet("doctor catalogs", flag.ContinueOnError)
jsonOut := fs.Bool("json", false, "print catalog diagnostics as JSON")
if code, ok := parseCommandFlags(fs, args); !ok {
return code
}
if fs.NArg() != 0 {
fmt.Fprintln(os.Stderr, "usage: reasonix doctor catalogs [--json]")
return 2
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
items := []catalogInspection{
{Name: "sessions", Info: projectiondb.Inspect(ctx, sessioncatalog.DefaultPath())},
}
for _, command := range catalogCommands {
items = append(items, catalogInspection{Name: command.name, Info: projectiondb.Inspect(ctx, command.path())})
}
if *jsonOut {
enc := json.NewEncoder(os.Stdout)
enc.SetIndent("", " ")
if err := enc.Encode(items); err != nil {
fmt.Fprintln(os.Stderr, err)
return 1
}
return 0
}
fmt.Println("Reasonix disposable catalogs")
for _, item := range items {
state := "missing"
if item.Info.Exists && item.Info.Error == "" {
state = item.Info.Integrity
} else if item.Info.Error != "" {
state = "error: " + item.Info.Error
}
fmt.Printf(" %-8s %-12s schema=%d size=%d path=%s\n", item.Name, state, item.Info.Schema, item.Info.Size, item.Info.Path)
}
return 0
}
func catalogsCommand(args []string) int {
if len(args) < 2 || args[0] != "reindex" {
fmt.Fprintln(os.Stderr, "usage: reasonix catalogs reindex <catalog> [options]")
return 2
}
for _, command := range catalogCommands {
if args[1] == command.name {
return command.reindex(args[2:])
}
}
fmt.Fprintln(os.Stderr, "unknown catalog:", args[1])
return 2
}
func printCatalogStatus(status any, jsonOut bool) int {
if jsonOut {
enc := json.NewEncoder(os.Stdout)
enc.SetIndent("", " ")
if err := enc.Encode(status); err != nil {
fmt.Fprintln(os.Stderr, err)
return 1
}
return 0
}
b, _ := json.Marshal(status)
fmt.Println(string(b))
return 0
}
+58
View File
@@ -0,0 +1,58 @@
package cli
import (
"context"
"flag"
"fmt"
"os"
"path/filepath"
"reasonix/internal/config"
"reasonix/internal/historycatalog"
)
func init() {
registerCatalogCommand(catalogCommand{
name: "history", path: historycatalog.DefaultPath, reindex: reindexHistoryCatalog,
completionFlags: []cliCompletionFlag{
completionFlag("--json", cliCompletionNoValue),
completionFlag("--dir", cliCompletionPathValue),
},
})
}
func reindexHistoryCatalog(args []string) int {
fs := flag.NewFlagSet("catalogs reindex history", flag.ContinueOnError)
var dirs stringListFlag
jsonOut := fs.Bool("json", false, "print status as JSON")
fs.Var(&dirs, "dir", "session directory to index; repeat for multiple directories")
if code, ok := parseCommandFlags(fs, args); !ok {
return code
}
ctx := context.Background()
catalog, err := historycatalog.Open(ctx, historycatalog.Options{})
if err != nil {
fmt.Fprintln(os.Stderr, "error:", err)
return 1
}
defer catalog.Close(context.Background())
roots := []historycatalog.Root{}
if len(dirs) > 0 {
for _, dir := range dirs {
roots = append(roots, historycatalog.Root{Path: filepath.Clean(dir), Source: "global", Scope: "global"})
}
} else {
for _, target := range defaultSessionCatalogTargets() {
roots = append(roots, historycatalog.Root{Path: target.Path, Source: target.Scope, Scope: target.Scope, WorkspaceRoot: target.WorkspaceRoot})
roots = append(roots, historycatalog.Root{Path: filepath.Join(target.Path, "subagents"), Source: target.Scope, Scope: target.Scope, WorkspaceRoot: target.WorkspaceRoot, Subagents: true})
}
roots = append(roots, historycatalog.Root{Path: config.ArchiveDir(), Source: "archive", Scope: "global", Archive: true})
}
for _, root := range roots {
if err := catalog.ReconcileRoot(ctx, root); err != nil {
fmt.Fprintln(os.Stderr, "error:", err)
return 1
}
}
return printCatalogStatus(catalog.Status(), *jsonOut)
}
+2 -3
View File
@@ -164,9 +164,8 @@ func RunWithBuildInfo(args []string, info BuildInfo) int {
case "report":
configureCLIThemeFromConfig()
return reportCommand(rest)
case "session", "sessions":
configureCLIThemeFromConfig()
return sessionOrSessionsCommand(cmd, rest)
case "session", "sessions", "catalogs":
return runSessionOrCatalogCommand(cmd, rest)
case "hook", "hooks":
configureCLIThemeFromConfig()
return hookCommand(rest)
+3
View File
@@ -14,6 +14,9 @@ import (
)
func doctorCommand(args []string, version string) int {
if len(args) > 0 && args[0] == "catalogs" {
return doctorCatalogsCommand(args[1:])
}
if len(args) > 0 && args[0] == "sessions" {
return doctorSessionsCommand(args[1:])
}
+10 -1
View File
@@ -46,6 +46,15 @@ func completionSpecWithAliases(name string, aliases []string, flags []cliComplet
return cliCompletionSpec{name: name, aliases: aliases, flags: flags, subcommands: subcommands}
}
func catalogCompletionSpec(help cliCompletionFlag) cliCompletionSpec {
reindex := make([]cliCompletionSpec, 0, len(catalogCommands))
for _, command := range catalogCommands {
flags := append(slices.Clone(command.completionFlags), help)
reindex = append(reindex, completionSpec(command.name, flags))
}
return completionSpec("catalogs", []cliCompletionFlag{help}, completionSpec("reindex", []cliCompletionFlag{help}, reindex...))
}
func cliCompletionRootSpec() cliCompletionSpec {
profile := completionFlag("--profile", cliCompletionStaticValue, "economy", "balanced", "delivery")
model := completionFlag("--model", cliCompletionModelValue)
@@ -89,7 +98,6 @@ func cliCompletionRootSpec() cliCompletionSpec {
completionFlag("--ablate", cliCompletionStaticValue, "none", "all", "evidence", "planner", "subagent", "retrieval", "compaction"),
help,
}
root := cliCompletionSpec{name: "reasonix", flags: append([]cliCompletionFlag{
model,
completionFlag("--max-steps", cliCompletionStaticValue),
@@ -207,6 +215,7 @@ func cliCompletionRootSpec() cliCompletionSpec {
completionFlag("--json", cliCompletionNoValue), completionFlag("--dir", cliCompletionPathValue), help,
}),
),
catalogCompletionSpec(help),
completionSpec("report", []cliCompletionFlag{help},
completionSpec("list", []cliCompletionFlag{help}),
completionSpec("show", []cliCompletionFlag{help}),
+287
View File
@@ -0,0 +1,287 @@
package history
import (
"context"
"path/filepath"
"sort"
"strings"
"sync"
"reasonix/internal/agent"
"reasonix/internal/historycatalog"
"reasonix/internal/provider"
"reasonix/internal/retrieval"
)
type indexedCatalogManager struct {
once sync.Once
mu sync.RWMutex
catalog *historycatalog.Catalog
roots map[string]historycatalog.Root
observers []func(historycatalog.Status, []string, string)
}
var processHistoryCatalog indexedCatalogManager
// RegisterCatalogRoots lets a host such as Desktop seed every saved project
// without constructing a controller. Opening and scanning remain asynchronous.
func RegisterCatalogRoots(roots []historycatalog.Root) { processHistoryCatalog.register(roots) }
// RegisterCatalogObserver subscribes a host to revision/progress changes. The
// callback payload contains roots and counters only, never query or content.
func RegisterCatalogObserver(observer func(historycatalog.Status, []string, string)) {
if observer == nil {
return
}
processHistoryCatalog.mu.Lock()
processHistoryCatalog.observers = append(processHistoryCatalog.observers, observer)
processHistoryCatalog.mu.Unlock()
}
// SharedCatalog returns the process projection when opening has completed.
// Nil means callers should return an explicit partial/opening response.
func SharedCatalog() *historycatalog.Catalog { return processHistoryCatalog.get() }
func FlushSharedCatalog(ctx context.Context) error {
if catalog := processHistoryCatalog.get(); catalog != nil {
return catalog.Flush(ctx)
}
return nil
}
func (m *indexedCatalogManager) register(roots []historycatalog.Root) {
m.mu.Lock()
if m.roots == nil {
m.roots = map[string]historycatalog.Root{}
}
for _, root := range roots {
if strings.TrimSpace(root.Path) != "" {
root.Path = filepath.Clean(root.Path)
m.roots[root.Path] = root
}
}
catalog := m.catalog
m.mu.Unlock()
if catalog != nil {
for _, root := range roots {
catalog.RegisterRoot(root)
}
}
m.once.Do(func() {
go func() {
catalog, err := historycatalog.Open(context.Background(), historycatalog.Options{OnRevision: processHistoryCatalog.publish})
if err != nil {
return
}
m.mu.Lock()
m.catalog = catalog
pending := make([]historycatalog.Root, 0, len(m.roots))
for _, root := range m.roots {
pending = append(pending, root)
}
m.mu.Unlock()
for _, root := range pending {
catalog.RegisterRoot(root)
}
}()
})
}
func (m *indexedCatalogManager) publish(status historycatalog.Status, roots []string, reason string) {
m.mu.RLock()
observers := append([]func(historycatalog.Status, []string, string){}, m.observers...)
m.mu.RUnlock()
for _, observer := range observers {
observer(status, append([]string{}, roots...), reason)
}
}
func (m *indexedCatalogManager) get() *historycatalog.Catalog {
m.mu.RLock()
defer m.mu.RUnlock()
return m.catalog
}
// PersistObserver returns the process-wide non-blocking projection sink. The
// authoritative save has already released its path/file locks before this is
// called by agent.Session.
func PersistObserver() agent.SessionPersistObserver { return historyPersistObserver{} }
type historyPersistObserver struct{}
func (historyPersistObserver) EnqueueSessionPersist(event agent.SessionPersistEvent) bool {
catalog := processHistoryCatalog.get()
if catalog == nil {
return false
}
processHistoryCatalog.mu.RLock()
defer processHistoryCatalog.mu.RUnlock()
bestLength := -1
var selected historycatalog.Root
for _, root := range processHistoryCatalog.roots {
if underRoot(event.Path, root.Path) && len(root.Path) > bestLength {
selected = root
bestLength = len(root.Path)
}
}
if bestLength < 0 {
return false
}
if event.Removed {
go func() { _ = catalog.Purge(context.Background(), event.Path) }()
return true
}
return catalog.EnqueuePersist(selected, event)
}
type IndexedSearcher struct {
legacy *Searcher
roots []historycatalog.Root
}
func NewIndexedSearcher(opts Options) *IndexedSearcher {
legacy := NewSearcher(opts)
roots := []historycatalog.Root{}
add := func(path, source, scope, workspace string, archive bool) {
if strings.TrimSpace(path) == "" {
return
}
roots = append(roots, historycatalog.Root{Path: path, Source: source, Scope: scope, WorkspaceRoot: workspace, Archive: archive})
}
add(opts.SessionDir, scopeProject, scopeProject, opts.SessionDir, false)
add(subagentsDir(opts.SessionDir), scopeProject, scopeProject, opts.SessionDir, false)
if filepath.Clean(opts.GlobalSessionDir) != filepath.Clean(opts.SessionDir) {
add(opts.GlobalSessionDir, scopeGlobal, scopeGlobal, "", false)
add(subagentsDir(opts.GlobalSessionDir), scopeGlobal, scopeGlobal, "", false)
}
add(opts.ArchiveDir, "archive", scopeGlobal, "", true)
processHistoryCatalog.register(roots)
return &IndexedSearcher{legacy: legacy, roots: roots}
}
func (s *IndexedSearcher) rootsFor(scope string) []string {
out := []string{}
for _, root := range s.roots {
if scope == scopeProject && root.Scope != scopeProject {
continue
}
out = append(out, root.Path)
}
return out
}
func (s *IndexedSearcher) Search(ctx context.Context, req SearchRequest) ([]Hit, error) {
query := strings.TrimSpace(req.Query)
if query == "" {
return nil, contextError("query is required")
}
queryTerms, err := retrieval.QueryTerms(query)
if err != nil {
return nil, err
}
scope, err := normalizeScope(req.Scope)
if err != nil {
return nil, err
}
limit := clamp(req.Limit, defaultLimit, maxLimit)
kinds, err := normalizeKinds(req.Kinds)
if err != nil {
return nil, err
}
catalog := processHistoryCatalog.get()
if catalog == nil {
return []Hit{}, nil
}
kindNames := make([]string, 0, len(kinds))
for kind := range kinds {
kindNames = append(kindNames, string(kind))
}
sort.Strings(kindNames)
result, err := catalog.Search(ctx, historycatalog.SearchRequest{
// Exact roots are the agent authority boundary. Catalog scope describes
// desktop grouping and must not change the history tool's project meaning.
Query: query,
Kinds: kindNames, ToolName: strings.TrimSpace(req.ToolName), Limit: min(limit*4, historycatalog.MaxLimit),
Roots: s.rootsFor(scope),
})
if err != nil {
return nil, err
}
loaded := map[string][]provider.Message{}
failed := map[string]bool{}
hits := make([]Hit, 0, len(result.Items))
for _, candidate := range result.Items {
if failed[candidate.SessionPath] {
continue
}
messages, ok := loaded[candidate.SessionPath]
if !ok {
state, known, identityErr := agent.SessionContentIdentity(candidate.SessionPath)
if identityErr != nil || (known && state.DigestHex != candidate.ContentDigest) {
failed[candidate.SessionPath] = true
catalog.EnqueueExisting(context.Background(), candidate.SessionPath)
continue
}
messages, err = loadMessages(candidate.SessionPath)
if err != nil {
failed[candidate.SessionPath] = true
continue
}
loaded[candidate.SessionPath] = messages
}
text, ok := candidateText(messages, candidate)
if !ok {
catalog.EnqueueExisting(context.Background(), candidate.SessionPath)
continue
}
hits = append(hits, Hit{Score: candidate.Score, SessionPath: candidate.SessionPath,
SessionID: sessionID(candidate.SessionPath), Source: candidate.Source, MessageIndex: candidate.MessageIndex,
Role: provider.Role(candidate.Role), Kind: Kind(candidate.Kind), ToolName: candidate.ToolName,
Snippet: retrieval.MakeSnippet(text, query, queryTerms, maxSnippet)})
}
hits = retrieval.KeepTopRelativeScore(hits, scoreFloor, func(hit Hit) float64 { return hit.Score })
if len(hits) > limit {
hits = hits[:limit]
}
return hits, nil
}
func candidateText(messages []provider.Message, candidate historycatalog.Candidate) (string, bool) {
if candidate.MessageIndex < 0 || candidate.MessageIndex >= len(messages) {
return "", false
}
msg := messages[candidate.MessageIndex]
switch Kind(candidate.Kind) {
case KindUserText:
return stripComposePrefixes(msg.Content), msg.Role == provider.RoleUser
case KindAssistantText:
return msg.Content, msg.Role == provider.RoleAssistant
case KindToolInput:
if candidate.PartIndex < 0 || candidate.PartIndex >= len(msg.ToolCalls) {
return "", false
}
call := msg.ToolCalls[candidate.PartIndex]
return strings.TrimSpace(call.Name + " " + call.Arguments), true
case KindToolError, KindToolOutput:
return strings.TrimSpace(msg.Name + " " + msg.Content), msg.Role == provider.RoleTool
default:
return "", false
}
}
func (s *IndexedSearcher) Around(ctx context.Context, req AroundRequest) ([]MessageContext, error) {
return s.legacy.Around(ctx, req)
}
func (s *IndexedSearcher) IndexStatus() historycatalog.Status {
if catalog := processHistoryCatalog.get(); catalog != nil {
return catalog.Status()
}
return historycatalog.Status{State: "opening", Pending: 1}
}
func contextError(message string) error { return &indexedInputError{message: message} }
type indexedInputError struct{ message string }
func (e *indexedInputError) Error() string { return e.message }
+22 -2
View File
@@ -6,11 +6,17 @@ import (
"fmt"
"strings"
"reasonix/internal/historycatalog"
"reasonix/internal/tool"
)
type historyTool struct {
searcher *Searcher
searcher historySearcher
}
type historySearcher interface {
Search(context.Context, SearchRequest) ([]Hit, error)
Around(context.Context, AroundRequest) ([]MessageContext, error)
}
// NewTool returns a read-only history retrieval tool bound to local sessions.
@@ -18,6 +24,13 @@ func NewTool(opts Options) tool.Tool {
return historyTool{searcher: NewSearcher(opts)}
}
// NewIndexedTool is the production history tool. It never performs a
// synchronous directory scan; while the shared projection is building it
// returns the already indexed subset and an explicit progress notice.
func NewIndexedTool(opts Options) tool.Tool {
return historyTool{searcher: NewIndexedSearcher(opts)}
}
func (historyTool) Name() string { return "history" }
func (historyTool) Description() string {
@@ -73,7 +86,14 @@ func (t historyTool) Execute(ctx context.Context, args json.RawMessage) (string,
if err != nil {
return "", err
}
return formatHits(in.Query, hits), nil
formatted := formatHits(in.Query, hits)
if statusProvider, ok := t.searcher.(interface{ IndexStatus() historycatalog.Status }); ok {
status := statusProvider.IndexStatus()
if status.State != "ready" || status.Pending > 0 || (status.Total > 0 && status.Indexed < status.Total) {
formatted = fmt.Sprintf("History index is still building (%d/%d); results may be incomplete.\n\n%s", status.Indexed, status.Total, formatted)
}
}
return formatted, nil
case "around":
if in.MessageIndex == nil {
return "", fmt.Errorf("message_index is required for operation=around")
+45
View File
@@ -0,0 +1,45 @@
package historycatalog
import (
"context"
"fmt"
"testing"
)
func BenchmarkWarmSearchTenThousandSources(b *testing.B) {
ctx := context.Background()
catalog, err := Open(ctx, Options{InMemory: true})
if err != nil {
b.Fatal(err)
}
b.Cleanup(func() { _ = catalog.Close(context.Background()) })
tx, err := catalog.db.Begin()
if err != nil {
b.Fatal(err)
}
for i := range 10_000 {
path := fmt.Sprintf("/sessions/session-%05d.jsonl", i)
if _, err := tx.Exec(`INSERT INTO history_sources(path,root,source,scope,workspace_root,health) VALUES(?,?,?,?,?,'ok')`,
path, "/sessions", "global", "global", ""); err != nil {
b.Fatal(err)
}
result, err := tx.Exec(`INSERT INTO history_documents(source_path,message_index,part_index,role,kind,tool_name,token_count)
VALUES(?,0,0,'user','user_text','',3)`, path)
if err != nil {
b.Fatal(err)
}
rowID, _ := result.LastInsertId()
if _, err := tx.Exec(`INSERT INTO history_fts(rowid,terms) VALUES(?,?)`, rowID, fmt.Sprintf("shared marker item%d", i)); err != nil {
b.Fatal(err)
}
}
if err := tx.Commit(); err != nil {
b.Fatal(err)
}
b.ResetTimer()
for range b.N {
if _, err := catalog.Search(ctx, SearchRequest{Query: "shared marker", Roots: []string{"/sessions"}, Limit: 50}); err != nil {
b.Fatal(err)
}
}
}
+744
View File
@@ -0,0 +1,744 @@
package historycatalog
import (
"context"
"crypto/sha256"
"database/sql"
"encoding/hex"
"errors"
"fmt"
"os"
"path/filepath"
"runtime"
"sort"
"strings"
"sync"
"sync/atomic"
"time"
"reasonix/internal/agent"
"reasonix/internal/projectiondb"
"reasonix/internal/provider"
"reasonix/internal/retrieval"
"reasonix/internal/store"
)
const defaultMissingGrace = 30 * time.Second
type Catalog struct {
db *sql.DB
opts Options
revision atomic.Uint64
statusMu sync.RWMutex
status Status
ctx context.Context
cancel context.CancelFunc
queue chan string
rootCh chan string
flushCh chan chan struct{}
mu sync.Mutex
paths map[string]queuedPath
roots map[string]Root
dirtyRoots map[string]bool
wg sync.WaitGroup
}
type queuedPath struct {
root Root
appendFrom int
}
func Open(ctx context.Context, opts Options) (*Catalog, error) {
if opts.Path == "" {
opts.Path = DefaultPath()
}
if opts.Now == nil {
opts.Now = time.Now
}
if opts.QueueCapacity <= 0 {
opts.QueueCapacity = 1024
}
if opts.MissingGrace <= 0 {
opts.MissingGrace = defaultMissingGrace
}
if opts.ReconcileInterval <= 0 {
opts.ReconcileInterval = time.Minute
}
handle, err := projectiondb.Open(ctx, projectiondb.OpenOptions{
Path: opts.Path, MemoryName: "history-search", Migrations: migrations(), InMemory: opts.InMemory,
MaxOpenConns: 4, Now: opts.Now, SecureDelete: true, AutoVacuum: true,
})
if err != nil {
return nil, err
}
workerCtx, cancel := context.WithCancel(context.Background())
c := &Catalog{db: handle.DB, opts: opts, ctx: workerCtx, cancel: cancel,
queue: make(chan string, opts.QueueCapacity), rootCh: make(chan string, 64),
flushCh: make(chan chan struct{}, 1),
paths: map[string]queuedPath{}, roots: map[string]Root{}, dirtyRoots: map[string]bool{},
status: Status{State: string(handle.Status.State), Mode: handle.Status.Mode, Path: handle.Status.Path,
LastError: handle.Status.LastError, QuarantinedPath: handle.Status.QuarantinedPath}}
if err := c.db.QueryRowContext(ctx, `SELECT revision FROM history_state WHERE id=1`).Scan(new(uint64)); err != nil {
_ = c.db.Close()
cancel()
return nil, err
}
if err := c.ensureTokenizerVersion(ctx); err != nil {
_ = c.db.Close()
cancel()
return nil, err
}
var revision uint64
_ = c.db.QueryRowContext(ctx, `SELECT revision FROM history_state WHERE id=1`).Scan(&revision)
c.revision.Store(revision)
c.refreshStatus(ctx)
c.wg.Add(1)
go c.worker()
return c, nil
}
func (c *Catalog) ensureTokenizerVersion(ctx context.Context) error {
var version int
if err := c.db.QueryRowContext(ctx, `SELECT tokenizer_version FROM history_state WHERE id=1`).Scan(&version); err != nil {
return err
}
if version == TokenizerVersion {
return nil
}
tx, err := c.db.BeginTx(ctx, nil)
if err != nil {
return err
}
for _, statement := range []string{`DELETE FROM history_fts`, `DELETE FROM history_documents`,
`DELETE FROM history_sources`, `DELETE FROM history_roots`} {
if _, err := tx.ExecContext(ctx, statement); err != nil {
_ = tx.Rollback()
return err
}
}
if _, err := tx.ExecContext(ctx, `UPDATE history_state SET tokenizer_version=?,revision=revision+1 WHERE id=1`, TokenizerVersion); err != nil {
_ = tx.Rollback()
return err
}
return tx.Commit()
}
func (c *Catalog) RegisterRoot(root Root) bool {
if c == nil || strings.TrimSpace(root.Path) == "" {
return false
}
root.Path = filepath.Clean(root.Path)
if root.Scope != "project" {
root.Scope = "global"
root.WorkspaceRoot = ""
}
c.mu.Lock()
_, alreadyRegistered := c.roots[root.Path]
c.roots[root.Path] = root
c.mu.Unlock()
if !alreadyRegistered {
c.statusMu.Lock()
c.status.Pending++
c.statusMu.Unlock()
}
select {
case c.rootCh <- root.Path:
return true
default:
c.markRootDirty(root.Path)
return false
}
}
// ReconcileRoot performs one deterministic scan. Production callers normally
// use RegisterRoot; the synchronous form exists for doctor/reindex and tests.
func (c *Catalog) ReconcileRoot(ctx context.Context, root Root) error {
root.Path = filepath.Clean(root.Path)
if root.Scope != "project" {
root.Scope = "global"
root.WorkspaceRoot = ""
}
return c.reconcileRoot(ctx, root)
}
func (c *Catalog) EnqueuePath(root Root, path string) bool {
return c.enqueuePath(root, path, -1)
}
func (c *Catalog) EnqueuePersist(root Root, event agent.SessionPersistEvent) bool {
appendFrom := event.AppendFrom
if event.Rewrite {
appendFrom = -1
}
return c.enqueuePath(root, event.Path, appendFrom)
}
func (c *Catalog) enqueuePath(root Root, path string, appendFrom int) bool {
if c == nil || strings.TrimSpace(path) == "" {
return false
}
path = filepath.Clean(path)
c.mu.Lock()
if queued, exists := c.paths[path]; exists {
queued.root = root
if queued.appendFrom < 0 || appendFrom < 0 {
queued.appendFrom = -1
} else if appendFrom < queued.appendFrom {
queued.appendFrom = appendFrom
}
c.paths[path] = queued
c.mu.Unlock()
return true
}
c.paths[path] = queuedPath{root: root, appendFrom: appendFrom}
c.mu.Unlock()
select {
case c.queue <- path:
return true
default:
c.mu.Lock()
delete(c.paths, path)
c.dirtyRoots[root.Path] = true
c.mu.Unlock()
return false
}
}
// EnqueueExisting prioritizes a source already known to the catalog without
// requiring the caller to retain its root metadata.
func (c *Catalog) EnqueueExisting(ctx context.Context, path string) bool {
var root Root
err := c.db.QueryRowContext(ctx, `SELECT root,source,scope,workspace_root FROM history_sources WHERE path=?`, filepath.Clean(path)).Scan(
&root.Path, &root.Source, &root.Scope, &root.WorkspaceRoot)
if err != nil {
return false
}
return c.EnqueuePath(root, path)
}
func (c *Catalog) worker() {
defer c.wg.Done()
ticker := time.NewTicker(c.opts.ReconcileInterval)
defer ticker.Stop()
for {
if root, ok := c.takeDirtyRoot(); ok {
_ = c.reconcileRoot(c.ctx, root)
continue
}
select {
case <-c.ctx.Done():
return
case <-ticker.C:
c.markAllRootsDirty()
case done := <-c.flushCh:
close(done)
case path := <-c.queue:
c.mu.Lock()
queued := c.paths[path]
delete(c.paths, path)
c.mu.Unlock()
_ = c.indexPath(c.ctx, queued.root, path, 0, queued.appendFrom)
case path := <-c.rootCh:
c.mu.Lock()
root, ok := c.roots[path]
c.mu.Unlock()
if ok {
_ = c.reconcileRoot(c.ctx, root)
}
}
}
}
func (c *Catalog) Flush(ctx context.Context) error {
if c == nil {
return nil
}
done := make(chan struct{})
select {
case c.flushCh <- done:
case <-ctx.Done():
return ctx.Err()
}
select {
case <-done:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
func (c *Catalog) markRootDirty(path string) {
c.mu.Lock()
c.dirtyRoots[path] = true
c.mu.Unlock()
}
func (c *Catalog) markAllRootsDirty() {
c.mu.Lock()
for path := range c.roots {
c.dirtyRoots[path] = true
}
c.mu.Unlock()
}
func (c *Catalog) takeDirtyRoot() (Root, bool) {
c.mu.Lock()
defer c.mu.Unlock()
for path := range c.dirtyRoots {
delete(c.dirtyRoots, path)
root, ok := c.roots[path]
return root, ok
}
return Root{}, false
}
func (c *Catalog) reconcileRoot(ctx context.Context, root Root) error {
entries, err := os.ReadDir(root.Path)
if err != nil && !errors.Is(err, os.ErrNotExist) {
c.setError(err)
return err
}
paths := make([]string, 0, len(entries))
for _, entry := range entries {
if entry.IsDir() || !store.IsSessionTranscriptName(entry.Name()) {
continue
}
path := filepath.Join(root.Path, entry.Name())
if !root.Archive && !agent.IsVisibleSession(path) {
continue
}
paths = append(paths, path)
}
sort.Strings(paths)
now := c.opts.Now().UnixMilli()
tx, err := c.db.BeginTx(ctx, nil)
if err != nil {
return err
}
var generation int64
if err := tx.QueryRowContext(ctx, `INSERT INTO history_roots(path,source,scope,workspace_root,scan_generation,state,total)
VALUES(?,?,?,?,1,'scanning',?) ON CONFLICT(path) DO UPDATE SET source=excluded.source,scope=excluded.scope,
workspace_root=excluded.workspace_root,scan_generation=history_roots.scan_generation+1,state='scanning',error='',total=excluded.total
RETURNING scan_generation`, root.Path, root.Source, root.Scope, root.WorkspaceRoot, len(paths)).Scan(&generation); err != nil {
_ = tx.Rollback()
return err
}
if err := tx.Commit(); err != nil {
return err
}
indexed := 0
for i, path := range paths {
if err := ctx.Err(); err != nil {
return err
}
if err := c.indexPath(ctx, root, path, generation, -1); err == nil {
indexed++
}
if (i+1)%32 == 0 {
runtime.Gosched()
}
}
tx, err = c.db.BeginTx(ctx, nil)
if err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `UPDATE history_sources SET missing_since=CASE WHEN missing_since=0 THEN ? ELSE missing_since END,
health='missing' WHERE root=? AND seen_generation<>?`, now, root.Path, generation); err != nil {
_ = tx.Rollback()
return err
}
cutoff := now - c.opts.MissingGrace.Milliseconds()
if _, err := tx.ExecContext(ctx, `DELETE FROM history_fts WHERE rowid IN (
SELECT d.id FROM history_documents d JOIN history_sources s ON s.path=d.source_path
WHERE s.root=? AND s.seen_generation<>? AND s.missing_since>0 AND s.missing_since<=?
)`, root.Path, generation, cutoff); err != nil {
_ = tx.Rollback()
return err
}
if _, err := tx.ExecContext(ctx, `DELETE FROM history_documents WHERE source_path IN (
SELECT path FROM history_sources WHERE root=? AND seen_generation<>? AND missing_since>0 AND missing_since<=?
)`, root.Path, generation, cutoff); err != nil {
_ = tx.Rollback()
return err
}
if _, err := tx.ExecContext(ctx, `DELETE FROM history_sources
WHERE root=? AND seen_generation<>? AND missing_since>0 AND missing_since<=?`, root.Path, generation, cutoff); err != nil {
_ = tx.Rollback()
return err
}
if _, err := tx.ExecContext(ctx, `UPDATE history_roots SET state='ready',indexed=?,completed_at=?,scan_cursor='' WHERE path=?`, indexed, now, root.Path); err != nil {
_ = tx.Rollback()
return err
}
revision, err := bump(ctx, tx)
if err != nil {
_ = tx.Rollback()
return err
}
if err := tx.Commit(); err != nil {
return err
}
c.publish(revision, []string{root.Path}, "reconcile")
return nil
}
type indexedDocument struct {
message int
part int
role string
kind string
tool string
terms string
count int
}
func documents(messages []provider.Message) []indexedDocument {
out := []indexedDocument{}
appendDoc := func(message, part int, role, kind, tool, text string) {
terms := retrieval.Tokens(strings.TrimSpace(text))
if len(terms) == 0 {
return
}
out = append(out, indexedDocument{message: message, part: part, role: role, kind: kind, tool: tool, terms: strings.Join(terms, " "), count: len(terms)})
}
for i, msg := range messages {
switch msg.Role {
case provider.RoleUser:
appendDoc(i, 0, string(msg.Role), "user_text", "", msg.Content)
case provider.RoleAssistant:
appendDoc(i, 0, string(msg.Role), "assistant_text", "", msg.Content)
for part, call := range msg.ToolCalls {
appendDoc(i, part, string(msg.Role), "tool_input", call.Name, call.Name+" "+call.Arguments)
}
case provider.RoleTool:
lower := strings.ToLower(strings.TrimSpace(msg.Content))
if strings.HasPrefix(lower, "error:") || strings.HasPrefix(lower, "blocked:") || strings.Contains(lower, "permission denied") {
appendDoc(i, 0, string(msg.Role), "tool_error", msg.Name, msg.Name+" "+msg.Content)
}
appendDoc(i, 0, string(msg.Role), "tool_output", msg.Name, msg.Name+" "+msg.Content)
}
}
return out
}
func fileFingerprint(path string) string {
info, err := os.Stat(path)
if err != nil {
return ""
}
return fmt.Sprintf("%d:%d", info.Size(), info.ModTime().UnixNano())
}
func (c *Catalog) indexPath(ctx context.Context, root Root, path string, generation int64, appendFrom int) error {
if !root.Archive && !agent.IsVisibleSession(path) {
return c.Purge(ctx, path)
}
contentFingerprint := fileFingerprint(path)
metaFingerprint := fileFingerprint(agent.BranchMetaPath(path))
state, known, identityErr := agent.SessionContentIdentity(path)
digest := ""
revision := int64(0)
if identityErr == nil && known {
digest, revision = state.DigestHex, state.Revision
}
var oldFingerprint, oldMetaFingerprint, oldDigest string
var oldGeneration, oldRevision int64
var oldMessageCount int
err := c.db.QueryRowContext(ctx, `SELECT content_fingerprint,meta_fingerprint,content_digest,seen_generation,
content_revision,indexed_message_count FROM history_sources WHERE path=?`, path).Scan(
&oldFingerprint, &oldMetaFingerprint, &oldDigest, &oldGeneration, &oldRevision, &oldMessageCount)
if sourceProjectionUnchanged(err, oldFingerprint, contentFingerprint, oldMetaFingerprint, metaFingerprint, oldDigest, digest) {
if generation != 0 && oldGeneration != generation {
_, _ = c.db.ExecContext(ctx, `UPDATE history_sources SET seen_generation=?,missing_since=0,health='ok' WHERE path=?`, generation, path)
}
return nil
}
if err != nil && !errors.Is(err, sql.ErrNoRows) {
return err
}
if err == nil && known && appendFrom >= 0 {
handled, appendErr := c.tryAppendPath(ctx, root, path, generation, appendFrom, oldMessageCount, oldRevision,
revision, digest, contentFingerprint, metaFingerprint)
if appendErr != nil {
return appendErr
}
if handled {
return nil
}
}
session, err := agent.LoadSession(path)
if err != nil {
_, _ = c.db.ExecContext(ctx, `INSERT INTO history_sources(path,root,source,scope,workspace_root,content_fingerprint,meta_fingerprint,health,last_error,seen_generation)
VALUES(?,?,?,?,?,?,?,'corrupt',?,?) ON CONFLICT(path) DO UPDATE SET health='corrupt',last_error=excluded.last_error,
content_fingerprint=excluded.content_fingerprint,meta_fingerprint=excluded.meta_fingerprint,seen_generation=excluded.seen_generation`,
path, root.Path, root.Source, root.Scope, root.WorkspaceRoot, contentFingerprint, metaFingerprint, err.Error(), generation)
c.setError(err)
return err
}
messages := session.Snapshot()
if digest == "" {
h := sha256.New()
for _, doc := range documents(messages) {
_, _ = h.Write([]byte(doc.terms))
_, _ = h.Write([]byte{0})
}
digest = hex.EncodeToString(h.Sum(nil))
}
meta, _, _ := agent.LoadBranchMeta(path)
lastActivity := max(int64(0), agent.SessionContentModTime(path).UnixMilli())
// Hide stale terms as soon as the authoritative fingerprint changes. Rows
// remain available for retry and are atomically replaced below.
if _, err := c.db.ExecContext(ctx, `UPDATE history_sources SET health='stale',last_error='' WHERE path=?`, path); err != nil {
return err
}
tx, err := c.db.BeginTx(ctx, nil)
if err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `DELETE FROM history_fts WHERE rowid IN (SELECT id FROM history_documents WHERE source_path=?)`, path); err != nil {
_ = tx.Rollback()
return err
}
if _, err := tx.ExecContext(ctx, `DELETE FROM history_documents WHERE source_path=?`, path); err != nil {
_ = tx.Rollback()
return err
}
_, err = tx.ExecContext(ctx, `INSERT INTO history_sources(path,root,source,scope,workspace_root,content_revision,content_digest,
content_fingerprint,meta_fingerprint,message_count,indexed_message_count,custom_title,topic_id,topic_title,preview,created_at,
last_activity_at,health,missing_since,seen_generation,last_error) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,'ok',0,?,'')
ON CONFLICT(path) DO UPDATE SET root=excluded.root,source=excluded.source,scope=excluded.scope,workspace_root=excluded.workspace_root,
content_revision=excluded.content_revision,content_digest=excluded.content_digest,content_fingerprint=excluded.content_fingerprint,
meta_fingerprint=excluded.meta_fingerprint,message_count=excluded.message_count,indexed_message_count=excluded.indexed_message_count,
custom_title=excluded.custom_title,topic_id=excluded.topic_id,topic_title=excluded.topic_title,preview=excluded.preview,
created_at=excluded.created_at,last_activity_at=excluded.last_activity_at,health='ok',missing_since=0,
seen_generation=excluded.seen_generation,last_error=''`, path, root.Path, root.Source, root.Scope, root.WorkspaceRoot, revision, digest,
contentFingerprint, metaFingerprint, len(messages), len(messages), meta.CustomTitle, meta.TopicID, meta.TopicTitle, meta.Preview,
meta.CreatedAt.UnixMilli(), lastActivity, generation)
if err != nil {
_ = tx.Rollback()
return err
}
for _, doc := range documents(messages) {
result, err := tx.ExecContext(ctx, `INSERT INTO history_documents(source_path,message_index,part_index,role,kind,tool_name,token_count)
VALUES(?,?,?,?,?,?,?)`, path, doc.message, doc.part, doc.role, doc.kind, doc.tool, doc.count)
if err != nil {
_ = tx.Rollback()
return err
}
rowID, err := result.LastInsertId()
if err != nil {
_ = tx.Rollback()
return err
}
if _, err := tx.ExecContext(ctx, `INSERT INTO history_fts(rowid,terms) VALUES(?,?)`, rowID, doc.terms); err != nil {
_ = tx.Rollback()
return err
}
}
newRevision, err := bump(ctx, tx)
if err != nil {
_ = tx.Rollback()
return err
}
if err := tx.Commit(); err != nil {
return err
}
c.publish(newRevision, []string{root.Path}, "source-indexed")
return nil
}
func bump(ctx context.Context, tx *sql.Tx) (uint64, error) {
if _, err := tx.ExecContext(ctx, `UPDATE history_state SET revision=revision+1 WHERE id=1`); err != nil {
return 0, err
}
var revision uint64
err := tx.QueryRowContext(ctx, `SELECT revision FROM history_state WHERE id=1`).Scan(&revision)
return revision, err
}
func (c *Catalog) Search(ctx context.Context, req SearchRequest) (SearchResult, error) {
out := SearchResult{Items: []Candidate{}, Revision: c.revision.Load(), Partial: c.Status().Pending > 0}
terms, err := retrieval.QueryTerms(req.Query)
if err != nil {
return out, err
}
limit := req.Limit
if limit <= 0 {
limit = DefaultLimit
}
if limit > MaxLimit {
limit = MaxLimit
}
match := make([]string, 0, len(terms))
for _, term := range terms {
match = append(match, `"`+strings.ReplaceAll(term, `"`, `""`)+`"`)
}
where := []string{`history_fts MATCH ?`, `s.health='ok'`, `s.missing_since=0`}
args := []any{strings.Join(match, " OR ")}
if req.Scope == "project" {
where = append(where, `s.scope='project'`, `s.workspace_root=?`)
args = append(args, strings.TrimSpace(req.WorkspaceRoot))
}
if len(req.Kinds) > 0 {
placeholders := make([]string, len(req.Kinds))
for i, kind := range req.Kinds {
placeholders[i] = "?"
args = append(args, kind)
}
where = append(where, `d.kind IN (`+strings.Join(placeholders, ",")+`)`)
}
if tool := strings.TrimSpace(req.ToolName); tool != "" {
where = append(where, `d.tool_name=?`)
args = append(args, tool)
}
if len(req.Roots) > 0 {
placeholders := make([]string, 0, len(req.Roots))
for _, root := range req.Roots {
if strings.TrimSpace(root) == "" {
continue
}
placeholders = append(placeholders, "?")
args = append(args, filepath.Clean(root))
}
if len(placeholders) > 0 {
where = append(where, `s.root IN (`+strings.Join(placeholders, ",")+`)`)
}
}
baseQuery := `SELECT d.id AS id,d.source_path AS source_path,s.root AS root,s.source AS source,s.scope AS scope,
s.workspace_root AS workspace_root,s.content_digest AS content_digest,d.message_index AS message_index,
d.part_index AS part_index,d.role AS role,d.kind AS kind,d.tool_name AS tool_name,bm25(history_fts) AS rank,
s.custom_title AS custom_title,s.topic_title AS topic_title,s.last_activity_at AS last_activity_at
FROM history_fts JOIN history_documents d ON d.id=history_fts.rowid JOIN history_sources s ON s.path=d.source_path
WHERE ` + strings.Join(where, ` AND `)
query := baseQuery + ` ORDER BY bm25(history_fts),d.source_path,d.message_index,d.part_index,d.id LIMIT ?`
if after := req.After; after != nil {
query = `WITH ranked AS MATERIALIZED (` + baseQuery + `)
SELECT * FROM ranked WHERE rank>? OR (rank=? AND source_path>?) OR
(rank=? AND source_path=? AND message_index>?) OR
(rank=? AND source_path=? AND message_index=? AND part_index>?) OR
(rank=? AND source_path=? AND message_index=? AND part_index=? AND id>?)
ORDER BY rank,source_path,message_index,part_index,id LIMIT ?`
args = append(args, after.Rank, after.Rank, after.SessionPath,
after.Rank, after.SessionPath, after.MessageIndex,
after.Rank, after.SessionPath, after.MessageIndex, after.PartIndex,
after.Rank, after.SessionPath, after.MessageIndex, after.PartIndex, after.RowID)
}
args = append(args, limit)
rows, err := c.db.QueryContext(ctx, query, args...)
if err != nil {
return out, err
}
defer rows.Close()
for rows.Next() {
var item Candidate
if err := rows.Scan(&item.RowID, &item.SessionPath, &item.Root, &item.Source, &item.Scope, &item.WorkspaceRoot, &item.ContentDigest,
&item.MessageIndex, &item.PartIndex, &item.Role, &item.Kind, &item.ToolName, &item.Rank,
&item.SessionTitle, &item.TopicTitle, &item.LastActivityAt); err != nil {
return out, err
}
if !catalogPathWithin(item.SessionPath, item.Root) {
continue
}
if item.Rank < 0 {
item.Score = -item.Rank
} else {
item.Score = 1 / (1 + item.Rank)
}
out.Items = append(out.Items, item)
}
return out, rows.Err()
}
func catalogPathWithin(path, root string) bool {
absPath, err := filepath.Abs(filepath.Clean(strings.TrimSpace(path)))
if err != nil {
return false
}
absRoot, err := filepath.Abs(filepath.Clean(strings.TrimSpace(root)))
if err != nil {
return false
}
rel, err := filepath.Rel(absRoot, absPath)
return err == nil && (rel == "." || (rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator))))
}
func (c *Catalog) Purge(ctx context.Context, path string) error {
if c == nil {
return nil
}
tx, err := c.db.BeginTx(ctx, nil)
if err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `DELETE FROM history_fts WHERE rowid IN (SELECT id FROM history_documents WHERE source_path=?)`, path); err != nil {
_ = tx.Rollback()
return err
}
if _, err := tx.ExecContext(ctx, `DELETE FROM history_sources WHERE path=?`, path); err != nil {
_ = tx.Rollback()
return err
}
revision, err := bump(ctx, tx)
if err != nil {
_ = tx.Rollback()
return err
}
if err := tx.Commit(); err != nil {
return err
}
_, _ = c.db.ExecContext(ctx, `PRAGMA wal_checkpoint(TRUNCATE)`)
_, _ = c.db.ExecContext(ctx, `PRAGMA incremental_vacuum(64)`)
c.publish(revision, nil, "purge")
return nil
}
func (c *Catalog) refreshStatus(ctx context.Context) {
var indexed, total, pending, failed int64
_ = c.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM history_sources WHERE health='ok'`).Scan(&indexed)
_ = c.db.QueryRowContext(ctx, `SELECT COALESCE(SUM(total),0) FROM history_roots`).Scan(&total)
_ = c.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM history_roots WHERE state<>'ready'`).Scan(&pending)
_ = c.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM history_sources WHERE health='corrupt'`).Scan(&failed)
c.statusMu.Lock()
c.status.Indexed, c.status.Total, c.status.Pending, c.status.Failed = indexed, total, pending, failed
c.status.Revision = c.revision.Load()
c.statusMu.Unlock()
}
func (c *Catalog) publish(revision uint64, roots []string, reason string) {
c.revision.Store(revision)
c.refreshStatus(context.Background())
if c.opts.OnRevision != nil {
c.opts.OnRevision(c.Status(), roots, reason)
}
}
func (c *Catalog) setError(err error) {
c.statusMu.Lock()
c.status.LastError = err.Error()
c.status.Failed++
c.statusMu.Unlock()
}
func (c *Catalog) Status() Status {
if c == nil {
return Status{State: "degraded", Mode: projectiondb.ModeMemory, LastError: "history catalog unavailable"}
}
c.statusMu.RLock()
defer c.statusMu.RUnlock()
return c.status
}
func (c *Catalog) Close(ctx context.Context) error {
if c == nil {
return nil
}
c.cancel()
done := make(chan struct{})
go func() { c.wg.Wait(); close(done) }()
select {
case <-done:
return c.db.Close()
case <-ctx.Done():
return ctx.Err()
}
}
+207
View File
@@ -0,0 +1,207 @@
package historycatalog
import (
"context"
"path/filepath"
"testing"
"reasonix/internal/agent"
"reasonix/internal/provider"
)
func saveMessages(t *testing.T, path string, messages ...provider.Message) {
t.Helper()
session := agent.NewSession("")
for _, message := range messages {
session.Add(message)
}
if err := session.Save(path); err != nil {
t.Fatal(err)
}
}
func TestReconcileAndSearchFTSWithoutStoredBody(t *testing.T) {
t.Parallel()
ctx := context.Background()
root := t.TempDir()
path := filepath.Join(root, "decision.jsonl")
saveMessages(t, path,
provider.Message{Role: provider.RoleUser, Content: "Should we use vector embeddings?"},
provider.Message{Role: provider.RoleAssistant, Content: "Keep lightweight BM25 retrieval for history."})
catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "history.sqlite")})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = catalog.Close(context.Background()) })
registered := Root{Path: root, Source: "project", Scope: "project", WorkspaceRoot: "/workspace"}
if err := catalog.ReconcileRoot(ctx, registered); err != nil {
t.Fatal(err)
}
result, err := catalog.Search(ctx, SearchRequest{Query: "lightweight BM25", Scope: "project", WorkspaceRoot: "/workspace", Kinds: []string{"assistant_text"}, Roots: []string{root}, Limit: 5})
if err != nil {
t.Fatal(err)
}
if len(result.Items) != 1 || result.Items[0].SessionPath != path || result.Items[0].MessageIndex != 1 {
t.Fatalf("result=%#v", result)
}
var storedTerms string
if err := catalog.db.QueryRow(`SELECT terms FROM history_fts WHERE rowid=?`, result.Items[0].RowID).Scan(&storedTerms); err == nil {
t.Fatalf("contentless FTS unexpectedly returned stored terms %q", storedTerms)
}
}
func TestRewriteRemovesOldTerms(t *testing.T) {
t.Parallel()
ctx := context.Background()
root := t.TempDir()
path := filepath.Join(root, "rewrite.jsonl")
saveMessages(t, path, provider.Message{Role: provider.RoleUser, Content: "obsolete unicorn marker"})
catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "history.sqlite")})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = catalog.Close(context.Background()) })
target := Root{Path: root, Source: "project", Scope: "project", WorkspaceRoot: root}
if err := catalog.ReconcileRoot(ctx, target); err != nil {
t.Fatal(err)
}
session, err := agent.LoadSession(path)
if err != nil {
t.Fatal(err)
}
session.Rewrite([]provider.Message{{Role: provider.RoleUser, Content: "replacement phoenix marker"}}, "test")
if err := session.SaveRewrite(path); err != nil {
t.Fatal(err)
}
if err := catalog.ReconcileRoot(ctx, target); err != nil {
t.Fatal(err)
}
oldResult, err := catalog.Search(ctx, SearchRequest{Query: "unicorn", Kinds: []string{"user_text"}, Roots: []string{root}})
if err != nil {
t.Fatal(err)
}
if len(oldResult.Items) != 0 {
t.Fatalf("stale terms survived rewrite: %#v", oldResult.Items)
}
newResult, err := catalog.Search(ctx, SearchRequest{Query: "phoenix", Kinds: []string{"user_text"}, Roots: []string{root}})
if err != nil || len(newResult.Items) != 1 {
t.Fatalf("newResult=%#v err=%v", newResult, err)
}
}
func TestAppendIndexesOnlyDisplayTail(t *testing.T) {
t.Parallel()
ctx := context.Background()
root := t.TempDir()
path := filepath.Join(root, "append.jsonl")
saveMessages(t, path, provider.Message{Role: provider.RoleUser, Content: "stable prefix marker"})
catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "history.sqlite")})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = catalog.Close(context.Background()) })
target := Root{Path: root, Source: "project", Scope: "project", WorkspaceRoot: root}
if err := catalog.ReconcileRoot(ctx, target); err != nil {
t.Fatal(err)
}
var prefixRowID int64
if err := catalog.db.QueryRow(`SELECT id FROM history_documents WHERE source_path=? AND message_index=0`, path).Scan(&prefixRowID); err != nil {
t.Fatal(err)
}
session, err := agent.LoadSession(path)
if err != nil {
t.Fatal(err)
}
session.Add(provider.Message{Role: provider.RoleAssistant, Content: "new append-only phoenix"})
if err := session.SaveSnapshot(path); err != nil {
t.Fatal(err)
}
if err := catalog.indexPath(ctx, target, path, 0, 1); err != nil {
t.Fatal(err)
}
var unchangedRowID int64
if err := catalog.db.QueryRow(`SELECT id FROM history_documents WHERE source_path=? AND message_index=0`, path).Scan(&unchangedRowID); err != nil {
t.Fatal(err)
}
if unchangedRowID != prefixRowID {
t.Fatalf("prefix row was rebuilt: before=%d after=%d", prefixRowID, unchangedRowID)
}
result, err := catalog.Search(ctx, SearchRequest{Query: "phoenix", Roots: []string{root}})
if err != nil || len(result.Items) != 1 || result.Items[0].MessageIndex != 1 {
t.Fatalf("appended result=%#v err=%v", result, err)
}
}
func TestSearchKeysetContinuesPastCatalogLimit(t *testing.T) {
t.Parallel()
ctx := context.Background()
root := t.TempDir()
path := filepath.Join(root, "many.jsonl")
messages := make([]provider.Message, 0, 5)
for range 5 {
messages = append(messages, provider.Message{Role: provider.RoleUser, Content: "shared pagination marker"})
}
saveMessages(t, path, messages...)
catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "history.sqlite")})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = catalog.Close(context.Background()) })
if err := catalog.ReconcileRoot(ctx, Root{Path: root, Source: "project", Scope: "project", WorkspaceRoot: root}); err != nil {
t.Fatal(err)
}
request := SearchRequest{Query: "pagination", Kinds: []string{"user_text"}, Roots: []string{root}, Limit: 2}
seen := map[int]bool{}
for {
result, err := catalog.Search(ctx, request)
if err != nil {
t.Fatal(err)
}
if len(result.Items) == 0 {
break
}
for _, item := range result.Items {
if seen[item.MessageIndex] {
t.Fatalf("message %d repeated across keyset pages", item.MessageIndex)
}
seen[item.MessageIndex] = true
}
last := result.Items[len(result.Items)-1]
request.After = &SearchCursor{Rank: last.Rank, SessionPath: last.SessionPath, MessageIndex: last.MessageIndex,
PartIndex: last.PartIndex, RowID: last.RowID}
}
if len(seen) != len(messages) {
t.Fatalf("indexed messages=%d, want %d", len(seen), len(messages))
}
}
func TestTokenizerVersionMismatchClearsMixedProjection(t *testing.T) {
t.Parallel()
ctx := context.Background()
root := t.TempDir()
path := filepath.Join(root, "old-tokenizer.jsonl")
databasePath := filepath.Join(t.TempDir(), "history.sqlite")
saveMessages(t, path, provider.Message{Role: provider.RoleUser, Content: "tokenizer migration marker"})
catalog, err := Open(ctx, Options{Path: databasePath})
if err != nil {
t.Fatal(err)
}
if err := catalog.ReconcileRoot(ctx, Root{Path: root, Source: "project", Scope: "project", WorkspaceRoot: root}); err != nil {
t.Fatal(err)
}
if _, err := catalog.db.Exec(`UPDATE history_state SET tokenizer_version=?`, TokenizerVersion+1); err != nil {
t.Fatal(err)
}
if err := catalog.Close(ctx); err != nil {
t.Fatal(err)
}
reopened, err := Open(ctx, Options{Path: databasePath})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = reopened.Close(context.Background()) })
result, err := reopened.Search(ctx, SearchRequest{Query: "migration", Roots: []string{root}})
if err != nil || len(result.Items) != 0 {
t.Fatalf("mixed tokenizer rows survived: result=%#v err=%v", result, err)
}
}
+76
View File
@@ -0,0 +1,76 @@
package historycatalog
import (
"context"
"reasonix/internal/agent"
"reasonix/internal/store"
)
func sourceProjectionUnchanged(queryErr error, oldContent, content, oldMeta, meta, oldDigest, digest string) bool {
return queryErr == nil && oldContent == content && oldMeta == meta && (digest == "" || digest == oldDigest)
}
func (c *Catalog) tryAppendPath(ctx context.Context, root Root, path string, generation int64, appendFrom,
oldMessageCount int, oldRevision, revision int64, digest, contentFingerprint, metaFingerprint string) (bool, error) {
if appendFrom != oldMessageCount || revision != oldRevision+1 {
return false, nil
}
index, err := agent.LoadSessionDisplayIndex(store.SessionDisplayIndex(path))
if err != nil || !index.RevisionKnown || index.Revision != revision || index.ContentDigest != digest || index.MessageCount < appendFrom {
return false, nil
}
tail, checkedIndex, err := agent.LoadSessionDisplayMessageRange(path, appendFrom, index.MessageCount)
if err != nil || checkedIndex.ContentDigest != digest || checkedIndex.Revision != revision {
return false, nil
}
meta, _, _ := agent.LoadBranchMeta(path)
lastActivity := max(int64(0), agent.SessionContentModTime(path).UnixMilli())
tx, err := c.db.BeginTx(ctx, nil)
if err != nil {
return false, err
}
result, err := tx.ExecContext(ctx, `UPDATE history_sources SET root=?,source=?,scope=?,workspace_root=?,content_revision=?,
content_digest=?,content_fingerprint=?,meta_fingerprint=?,message_count=?,indexed_message_count=?,custom_title=?,topic_id=?,
topic_title=?,preview=?,created_at=?,last_activity_at=?,health='ok',missing_since=0,
seen_generation=CASE WHEN ?>0 THEN ? ELSE seen_generation END,last_error=''
WHERE path=? AND content_revision=? AND indexed_message_count=?`, root.Path, root.Source, root.Scope, root.WorkspaceRoot,
revision, digest, contentFingerprint, metaFingerprint, index.MessageCount, index.MessageCount, meta.CustomTitle, meta.TopicID,
meta.TopicTitle, meta.Preview, meta.CreatedAt.UnixMilli(), lastActivity, generation, generation, path, oldRevision, oldMessageCount)
if err != nil {
_ = tx.Rollback()
return false, err
}
updated, _ := result.RowsAffected()
if updated != 1 {
_ = tx.Rollback()
return false, nil
}
for _, doc := range documents(tail) {
result, err := tx.ExecContext(ctx, `INSERT INTO history_documents(source_path,message_index,part_index,role,kind,tool_name,token_count)
VALUES(?,?,?,?,?,?,?)`, path, appendFrom+doc.message, doc.part, doc.role, doc.kind, doc.tool, doc.count)
if err != nil {
_ = tx.Rollback()
return false, err
}
rowID, err := result.LastInsertId()
if err != nil {
_ = tx.Rollback()
return false, err
}
if _, err := tx.ExecContext(ctx, `INSERT INTO history_fts(rowid,terms) VALUES(?,?)`, rowID, doc.terms); err != nil {
_ = tx.Rollback()
return false, err
}
}
newRevision, err := bump(ctx, tx)
if err != nil {
_ = tx.Rollback()
return false, err
}
if err := tx.Commit(); err != nil {
return false, err
}
c.publish(newRevision, []string{root.Path}, "source-appended")
return true, nil
}
+87
View File
@@ -0,0 +1,87 @@
package historycatalog
import (
"context"
"database/sql"
"reasonix/internal/projectiondb"
)
const migrationV1 = `
CREATE TABLE history_state(
id INTEGER PRIMARY KEY CHECK(id=1),
revision INTEGER NOT NULL DEFAULT 0,
tokenizer_version INTEGER NOT NULL
);
INSERT INTO history_state(id,revision,tokenizer_version) VALUES(1,0,1);
CREATE TABLE history_roots(
path TEXT PRIMARY KEY,
source TEXT NOT NULL,
scope TEXT NOT NULL,
workspace_root TEXT NOT NULL DEFAULT '',
signature TEXT NOT NULL DEFAULT '',
scan_generation INTEGER NOT NULL DEFAULT 0,
scan_cursor TEXT NOT NULL DEFAULT '',
state TEXT NOT NULL DEFAULT 'pending',
error TEXT NOT NULL DEFAULT '',
indexed INTEGER NOT NULL DEFAULT 0,
total INTEGER NOT NULL DEFAULT 0,
completed_at INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE history_sources(
path TEXT PRIMARY KEY,
root TEXT NOT NULL,
source TEXT NOT NULL,
scope TEXT NOT NULL,
workspace_root TEXT NOT NULL DEFAULT '',
content_revision INTEGER NOT NULL DEFAULT 0,
content_digest TEXT NOT NULL DEFAULT '',
content_fingerprint TEXT NOT NULL DEFAULT '',
meta_fingerprint TEXT NOT NULL DEFAULT '',
message_count INTEGER NOT NULL DEFAULT 0,
indexed_message_count INTEGER NOT NULL DEFAULT 0,
custom_title TEXT NOT NULL DEFAULT '',
topic_id TEXT NOT NULL DEFAULT '',
topic_title TEXT NOT NULL DEFAULT '',
preview TEXT NOT NULL DEFAULT '',
created_at INTEGER NOT NULL DEFAULT 0,
last_activity_at INTEGER NOT NULL DEFAULT 0,
health TEXT NOT NULL DEFAULT 'ok',
missing_since INTEGER NOT NULL DEFAULT 0,
seen_generation INTEGER NOT NULL DEFAULT 0,
last_error TEXT NOT NULL DEFAULT ''
);
CREATE TABLE history_documents(
id INTEGER PRIMARY KEY,
source_path TEXT NOT NULL REFERENCES history_sources(path) ON DELETE CASCADE,
message_index INTEGER NOT NULL,
part_index INTEGER NOT NULL DEFAULT 0,
role TEXT NOT NULL,
kind TEXT NOT NULL,
tool_name TEXT NOT NULL DEFAULT '',
token_count INTEGER NOT NULL DEFAULT 0,
UNIQUE(source_path,message_index,kind,part_index)
);
CREATE VIRTUAL TABLE history_fts USING fts5(
terms,
content='',
contentless_delete=1,
tokenize='unicode61 remove_diacritics 2'
);
CREATE INDEX idx_history_sources_scope_activity
ON history_sources(scope,workspace_root,last_activity_at DESC,path ASC);
CREATE INDEX idx_history_documents_source
ON history_documents(source_path,message_index,part_index,id);
`
func migrations() []projectiondb.Migration {
return []projectiondb.Migration{{Version: 1, Apply: func(ctx context.Context, tx *sql.Tx) error {
_, err := tx.ExecContext(ctx, migrationV1)
return err
}}}
}
+101
View File
@@ -0,0 +1,101 @@
// Package historycatalog maintains a disposable FTS projection of saved
// session history. Session JSONL/event/meta files remain authoritative.
package historycatalog
import (
"path/filepath"
"time"
"reasonix/internal/config"
"reasonix/internal/projectiondb"
)
const (
SchemaVersion = 1
TokenizerVersion = 1
DefaultLimit = 50
MaxLimit = 200
)
type Status struct {
State string `json:"state"`
Mode projectiondb.Mode `json:"mode"`
Path string `json:"path,omitempty"`
Revision uint64 `json:"revision"`
Indexed int64 `json:"indexed"`
Total int64 `json:"total"`
Pending int64 `json:"pending"`
Failed int64 `json:"failed"`
LastError string `json:"lastError,omitempty"`
QuarantinedPath string `json:"quarantinedPath,omitempty"`
}
type Options struct {
Path string
InMemory bool
QueueCapacity int
MissingGrace time.Duration
ReconcileInterval time.Duration
Now func() time.Time
OnRevision func(Status, []string, string)
}
type Root struct {
Path string
Source string
Scope string
WorkspaceRoot string
Subagents bool
Archive bool
}
type Candidate struct {
RowID int64
SessionPath string
Root string
Source string
Scope string
WorkspaceRoot string
ContentDigest string
MessageIndex int
PartIndex int
Role string
Kind string
ToolName string
Rank float64
Score float64
SessionTitle string
TopicTitle string
LastActivityAt int64
}
type SearchRequest struct {
Query string
Scope string
WorkspaceRoot string
Kinds []string
ToolName string
Limit int
Roots []string
After *SearchCursor
}
// SearchCursor is a stable keyset over SQLite's BM25 rank and deterministic
// tie-break keys. Lower BM25 ranks sort first.
type SearchCursor struct {
Rank float64
SessionPath string
MessageIndex int
PartIndex int
RowID int64
}
type SearchResult struct {
Items []Candidate
Revision uint64
Partial bool
}
func DefaultPath() string {
return filepath.Join(config.CacheDir(), "history-search", "v1.sqlite")
}
+6 -5
View File
@@ -94,11 +94,12 @@ func Open(ctx context.Context, opts Options) (*Catalog, error) {
status: Status{State: StateOpening, Path: opts.Path},
}
handle, err := projectiondb.Open(ctx, projectiondb.OpenOptions{
Path: opts.Path,
MemoryName: "session-catalog",
Migrations: sessionMigrations(),
InMemory: opts.InMemory,
Now: opts.Now,
Path: opts.Path,
MemoryName: "session-catalog",
Migrations: sessionMigrations(),
InMemory: opts.InMemory,
MaxOpenConns: 4,
Now: opts.Now,
})
if err != nil {
return nil, err
+35 -1
View File
@@ -221,11 +221,45 @@ func TestSchemaMigrationLedgerRecordsEveryVersion(t *testing.T) {
}
versions = append(versions, version)
}
if fmt.Sprint(versions) != "[1 2]" {
if fmt.Sprint(versions) != "[1 2 3]" {
t.Fatalf("schema migration ledger = %v", versions)
}
}
func TestListSessionsUsesRevisionBoundKeysetCursor(t *testing.T) {
t.Parallel()
ctx := context.Background()
catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "catalog.sqlite"), DisableRepair: true})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = catalog.Close(context.Background()) })
for i, name := range []string{"a", "b", "c"} {
if err := catalog.UpsertSession(ctx, SessionRecord{
Path: filepath.Join("/sessions", name+".jsonl"), Directory: "/sessions", Scope: "global",
TopicID: name, CustomTitle: "title " + name, LastActivityAt: int64(i + 1),
TurnsState: TurnsValid, Health: HealthOK,
}); err != nil {
t.Fatal(err)
}
}
first, err := catalog.ListSessions(ctx, SessionPageRequest{Scope: "all", Limit: 2})
if err != nil || len(first.Items) != 2 || first.NextCursor == "" {
t.Fatalf("first=%#v err=%v", first, err)
}
second, err := catalog.ListSessions(ctx, SessionPageRequest{Scope: "all", Limit: 2, Cursor: first.NextCursor})
if err != nil || len(second.Items) != 1 || second.Items[0].CustomTitle != "title a" {
t.Fatalf("second=%#v err=%v", second, err)
}
if err := catalog.UpsertSession(ctx, SessionRecord{Path: "/sessions/d.jsonl", Directory: "/sessions", Scope: "global", TopicID: "d", LastActivityAt: 4, TurnsState: TurnsValid, Health: HealthOK}); err != nil {
t.Fatal(err)
}
stale, err := catalog.ListSessions(ctx, SessionPageRequest{Scope: "all", Cursor: first.NextCursor})
if err != nil || !stale.StaleCursor || len(stale.Items) != 0 {
t.Fatalf("stale=%#v err=%v", stale, err)
}
}
func TestDirectWriteDuringScanIsNotMarkedMissing(t *testing.T) {
t.Parallel()
ctx := context.Background()
+9
View File
@@ -96,6 +96,11 @@ const migrationV2 = `
ALTER TABLE catalog_topics ADD COLUMN metadata_present INTEGER NOT NULL DEFAULT 0;
`
const migrationV3 = `
CREATE INDEX IF NOT EXISTS idx_catalog_sessions_history
ON catalog_sessions(scope, workspace_root, last_activity_at DESC, path ASC);
`
func sessionMigrations() []projectiondb.Migration {
return []projectiondb.Migration{
{Version: 1, Apply: func(ctx context.Context, tx *sql.Tx) error {
@@ -106,5 +111,9 @@ func sessionMigrations() []projectiondb.Migration {
_, err := tx.ExecContext(ctx, migrationV2)
return err
}},
{Version: 3, Apply: func(ctx context.Context, tx *sql.Tx) error {
_, err := tx.ExecContext(ctx, migrationV3)
return err
}},
}
}
+158
View File
@@ -0,0 +1,158 @@
package sessioncatalog
import (
"context"
"database/sql"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"path/filepath"
"strings"
"time"
)
type sessionPageCursor struct {
Revision uint64 `json:"r"`
Activity int64 `json:"a"`
Path string `json:"p"`
}
const sessionSelectColumns = `path,directory,scope,workspace_root,topic_id,topic_title,
custom_title,created_at,last_activity_at,preview,turns,turns_state,recovered,
recovery_reason,recovery_digest,parent_id,content_fingerprint,meta_fingerprint,
health,missing_since`
func scanSession(scanner interface{ Scan(...any) error }) (SessionRecord, error) {
var record SessionRecord
err := scanner.Scan(&record.Path, &record.Directory, &record.Scope, &record.WorkspaceRoot,
&record.TopicID, &record.TopicTitle, &record.CustomTitle, &record.CreatedAt,
&record.LastActivityAt, &record.Preview, &record.Turns, &record.TurnsState,
&record.Recovered, &record.RecoveryReason, &record.RecoveryDigest,
&record.ParentID, &record.ContentFingerprint, &record.MetaFingerprint,
&record.Health, &record.MissingSince)
return record, err
}
// ListSessions returns only catalog metadata. It never opens a transcript or
// sidecar and therefore remains safe on startup and UI pagination paths.
func (c *Catalog) ListSessions(ctx context.Context, req SessionPageRequest) (SessionPage, error) {
out := SessionPage{Items: []SessionRecord{}, Revision: c.revision.Load()}
if req.Limit <= 0 {
req.Limit = DefaultLimit
}
if req.Limit > MaxLimit {
req.Limit = MaxLimit
}
cursor, err := decodeSessionCursor(req.Cursor)
if err != nil {
return out, err
}
if cursor != nil && cursor.Revision != out.Revision {
out.StaleCursor = true
return out, nil
}
where := []string{`missing_since=0`, `health<>'missing'`}
args := []any{}
switch strings.ToLower(strings.TrimSpace(req.Scope)) {
case "", "all":
case "project":
where = append(where, `scope='project'`, `workspace_root=?`)
args = append(args, strings.TrimSpace(req.WorkspaceRoot))
case "global":
where = append(where, `scope='global'`)
default:
return out, fmt.Errorf("invalid session catalog scope %q", req.Scope)
}
if directory := strings.TrimSpace(req.Directory); directory != "" {
where = append(where, `directory=?`)
args = append(args, filepath.Clean(directory))
}
if query := strings.ToLower(strings.TrimSpace(req.Query)); query != "" {
where = append(where, `(lower(custom_title) LIKE ? OR lower(preview) LIKE ? OR lower(topic_title) LIKE ? OR lower(topic_id) LIKE ?)`)
like := "%" + query + "%"
args = append(args, like, like, like, like)
}
appendSessionTimeFilter(&where, &args, req.TimeFilter, c.opts.Now())
if cursor != nil {
where = append(where, `(last_activity_at<? OR (last_activity_at=? AND path>?))`)
args = append(args, cursor.Activity, cursor.Activity, cursor.Path)
}
args = append(args, req.Limit+1)
rows, err := c.db.QueryContext(ctx, `SELECT `+sessionSelectColumns+` FROM catalog_sessions WHERE `+
strings.Join(where, ` AND `)+` ORDER BY last_activity_at DESC,path ASC LIMIT ?`, args...)
if err != nil {
return out, err
}
defer rows.Close()
for rows.Next() {
record, err := scanSession(rows)
if err != nil {
return out, err
}
out.Items = append(out.Items, record)
}
if err := rows.Err(); err != nil {
return out, err
}
if len(out.Items) > req.Limit {
out.Items = out.Items[:req.Limit]
last := out.Items[len(out.Items)-1]
out.NextCursor = encodeSessionCursor(sessionPageCursor{Revision: out.Revision, Activity: last.LastActivityAt, Path: last.Path})
}
return out, nil
}
func appendSessionTimeFilter(where *[]string, args *[]any, filter string, now time.Time) {
value := strings.ToLower(strings.TrimSpace(filter))
startToday := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location())
switch value {
case "", "all":
case "today":
*where = append(*where, `last_activity_at>=?`)
*args = append(*args, startToday.UnixMilli())
case "yesterday":
*where = append(*where, `last_activity_at>=?`, `last_activity_at<?`)
*args = append(*args, startToday.AddDate(0, 0, -1).UnixMilli(), startToday.UnixMilli())
case "older":
*where = append(*where, `last_activity_at<?`)
*args = append(*args, startToday.AddDate(0, 0, -1).UnixMilli())
default:
if cutoff := timeFilterCutoff(value, now); cutoff > 0 {
*where = append(*where, `last_activity_at>=?`)
*args = append(*args, cutoff)
}
}
}
func (c *Catalog) GetSession(ctx context.Context, path string) (SessionRecord, bool, error) {
path = filepath.Clean(strings.TrimSpace(path))
if path == "." || path == "" {
return SessionRecord{}, false, nil
}
record, err := scanSession(c.db.QueryRowContext(ctx, `SELECT `+sessionSelectColumns+` FROM catalog_sessions WHERE path=?`, path))
if errors.Is(err, sql.ErrNoRows) {
return SessionRecord{}, false, nil
}
return record, err == nil, err
}
func encodeSessionCursor(cursor sessionPageCursor) string {
b, _ := json.Marshal(cursor)
return base64.RawURLEncoding.EncodeToString(b)
}
func decodeSessionCursor(encoded string) (*sessionPageCursor, error) {
if strings.TrimSpace(encoded) == "" {
return nil, nil
}
b, err := base64.RawURLEncoding.DecodeString(encoded)
if err != nil {
return nil, fmt.Errorf("invalid session catalog cursor: %w", err)
}
var cursor sessionPageCursor
if err := json.Unmarshal(b, &cursor); err != nil || cursor.Path == "" {
return nil, errors.New("invalid session catalog cursor")
}
return &cursor, nil
}
+18 -1
View File
@@ -12,7 +12,7 @@ import (
)
const (
SchemaVersion = 2
SchemaVersion = 3
DefaultLimit = 50
MaxLimit = 200
)
@@ -159,6 +159,23 @@ type TopicPage struct {
Revision uint64 `json:"revision"`
}
type SessionPageRequest struct {
Scope string `json:"scope"`
WorkspaceRoot string `json:"workspaceRoot,omitempty"`
Directory string `json:"-"`
Cursor string `json:"cursor,omitempty"`
Limit int `json:"limit,omitempty"`
Query string `json:"query,omitempty"`
TimeFilter string `json:"timeFilter,omitempty"`
}
type SessionPage struct {
Items []SessionRecord `json:"items"`
NextCursor string `json:"nextCursor,omitempty"`
Revision uint64 `json:"revision"`
StaleCursor bool `json:"staleCursor,omitempty"`
}
// DefaultPath returns the disposable session catalog path under CacheDir.
// When the OS cache directory is unavailable it returns "" so callers fall
// back to an in-memory projection instead of writing into the project tree.
+7 -7
View File
@@ -36,7 +36,7 @@
},
"desktop/app_test.go": {
"essay": 2,
"test-file-size": 10227
"test-file-size": 10217
},
"desktop/bot_bridge.go": {
"essay": 11
@@ -135,7 +135,7 @@
"file-size": 91
},
"desktop/frontend/src/components/ProjectTree.tsx": {
"file-size": 1587
"file-size": 1445
},
"desktop/frontend/src/components/RichComposerInput.tsx": {
"file-size": 141
@@ -322,7 +322,7 @@
},
"desktop/tabs_topic_test.go": {
"essay": 1,
"test-file-size": 3409
"test-file-size": 3227
},
"desktop/temphelper_test.go": {
"essay": 4
@@ -577,8 +577,8 @@
},
"internal/agent/save.go": {
"complexity": 49,
"essay": 113,
"file-size": 1544,
"essay": 112,
"file-size": 1522,
"function-size": 124
},
"internal/agent/save_test.go": {
@@ -924,7 +924,7 @@
"file-size": 126
},
"internal/cli/shell_completion.go": {
"function-size": 117
"function-size": 112
},
"internal/cli/skill_hooks.go": {
"essay": 1
@@ -1507,7 +1507,7 @@
"internal/pluginpkg/pluginpkg.go": {
"complexity": 1,
"essay": 2,
"file-size": 426
"file-size": 407
},
"internal/pluginpkg/pluginpkg_test.go": {
"test-file-size": 67