merge: adopt task catalog from #8219

Add the cross-project task catalog, Desktop task center APIs, and CLI
reindex surface on the shared multi-domain SQLite projection stack while
keeping history/usage registration and the hardened session foundation.
This commit is contained in:
SivanCola committed 2026-08-11 01:52:01 +08:00
commit 0d41f2ca46
34 files changed
+1773 -52

No files matched your search

+7 -1
View File
@@ -58,6 +58,7 @@ import (
"reasonix/internal/skill"
"reasonix/internal/stats"
"reasonix/internal/store"
"reasonix/internal/taskcatalog"
"reasonix/internal/taskmonitor"
"reasonix/internal/tool"
)
@@ -2407,6 +2408,7 @@ func (a *App) clearActiveSessionRuntime(tab *WorkspaceTab, oldCtrl control.Sessi
RequireKey: false,
AutoPricingCurrency: a.desktopAutoPricingCurrency(),
StatsSource: "desktop",
TaskStore: a.taskStore(),
Sink: newSink,
WorkspaceRoot: snap.workspaceRoot,
SessionDir: sessionDirForSnapshot(snap),
@@ -4506,6 +4508,7 @@ func (a *App) buildSessionRebindCandidate(
RequireKey: false,
AutoPricingCurrency: a.desktopAutoPricingCurrency(),
StatsSource: "desktop",
TaskStore: a.taskStore(),
Sink: a.desktopControllerSink(sink, cfg.Notifications),
WorkspaceRoot: root,
SessionDir: sessionDir,
@@ -10073,6 +10076,7 @@ func (a *App) SetModelForTab(tabID, name string) (retErr error) {
RequireKey: false,
AutoPricingCurrency: a.desktopAutoPricingCurrency(),
StatsSource: "desktop",
TaskStore: a.taskStore(),
Sink: snap.sink,
WorkspaceRoot: snap.workspaceRoot,
SessionDir: sessionDirForSnapshot(snap),
@@ -10256,6 +10260,7 @@ func (a *App) SetEffortForTab(tabID, level string) error {
RequireKey: false,
AutoPricingCurrency: a.desktopAutoPricingCurrency(),
StatsSource: "desktop",
TaskStore: a.taskStore(),
Sink: snap.sink,
WorkspaceRoot: snap.workspaceRoot,
SessionDir: sessionDirForSnapshot(snap),
@@ -10396,6 +10401,7 @@ func (a *App) SetTokenModeForTab(tabID, mode string) error {
RequireKey: false,
AutoPricingCurrency: a.desktopAutoPricingCurrency(),
StatsSource: "desktop",
TaskStore: a.taskStore(),
Sink: snap.sink,
WorkspaceRoot: snap.workspaceRoot,
SessionDir: sessionDirForSnapshot(snap),
@@ -11917,7 +11923,7 @@ func parseScope(s string) memory.Scope {
// taskStore is the Store backing the task monitor panel.
func (a *App) taskStore() taskmonitor.WriteStore {
return taskmonitor.NewFileStore(filepath.Join(".reasonix", "tasks"))
return taskcatalog.ObservedStore()
}
// taskControl returns the process-wide ControlService backing the task
@@ -81,7 +81,7 @@ const rawInitialBytes = [...initialJS, ...initialCSS, ...appShellCSS]
// Native Web Animations and frame-batched scrolling avoid an eager animation
// runtime. Goal request observability plus transcript scroll arbitration,
// 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);
// indexed-history bridge, and cross-project Task Center add small always-available
// contracts. Keep the raw allowance ratcheted while gzip stays flat.
assertBudget("initial raw JavaScript and CSS", rawInitialBytes, 2_250 * 1024);
assertBudget("largest initial JavaScript chunk raw", largestInitialJSRaw, 1_000 * 1024);
+7 -7
View File
@@ -1188,7 +1188,7 @@ export default function App() {
const sidebarWidth = useLayoutStore((s) => s.sidebarWidth);
const setSidebarWidth = useLayoutStore((s) => s.setSidebarWidth);
const [sidebarResizing, setSidebarResizing] = useState(false);
const [tasksOpen, setTasksOpen] = useState(false);
const [tasksOpen, setTasksOpen] = useState<false | "session" | "all">(false);
const [liveSidebarWidth, setLiveSidebarWidth] = useState<number | null>(null);
const [viewportWidth, setViewportWidth] = useState(() => (typeof window === "undefined" ? 1440 : window.innerWidth));
const [viewportHeight, setViewportHeight] = useState(() => (typeof window === "undefined" ? 720 : window.innerHeight));
@@ -3988,6 +3988,7 @@ export default function App() {
setSettingsTarget("models");
},
},
{ id: "cmd-task-center", group: t("palette.group.commands"), title: t("palette.cmd.taskCenter"), icon: <Activity size={15} />, compact: true, keywords: ["task", "tasks", "center", "任务", "任务中心"], run: () => setTasksOpen("all") },
{ id: "cmd-terminal", group: t("palette.group.commands"), title: t("rightDock.terminal"), icon: <TerminalSquare size={15} />, compact: true, keywords: ["terminal", "shell", "终端"], run: () => toggleTerminalPanel() },
{
id: "cmd-reload-runtime",
@@ -3999,8 +4000,7 @@ export default function App() {
run: () => {
const tabID = activeTab?.id;
if (!tabID) return;
// Success/queued feedback arrives as a tab notice from the Go side;
// only hard failures need a toast here.
// Success/queued feedback arrives as a tab notice; only hard failures need a toast.
void app.ReloadRuntime(tabID).catch((err) => showToast(err instanceof Error ? err.message : String(err), "error"));
},
},
@@ -4801,8 +4801,8 @@ export default function App() {
className={`topicbar__action-btn topicbar__action-btn--icon topicbar__action-btn--utility${tasksOpen ? " topicbar__action-btn--active" : ""}`}
type="button"
aria-label="Session summary"
aria-expanded={tasksOpen}
onClick={() => setTasksOpen((open) => !open)}
aria-expanded={Boolean(tasksOpen)}
onClick={() => setTasksOpen((open) => open ? false : "session")}
>
<Activity size={14} />
</button>
@@ -4811,9 +4811,9 @@ export default function App() {
<div className="taskmonitor-popover" role="dialog" aria-label="Session summary">
<Suspense fallback={null}>
<TaskMonitorPanel
key={`${activeTab?.id || activeTabId || "none"}:${activeTab?.workspaceRoot || "global"}:${activeTab?.sessionPath || ""}`}
key={`${activeTab?.id || activeTabId || "none"}:${activeTab?.workspaceRoot || "global"}:${activeTab?.sessionPath || ""}:${tasksOpen}`}
tabID={activeTab?.id || activeTabId || ""}
initialOpen
initialOpen initialScope={tasksOpen || "session"}
popover
summaryMode
onClose={() => setTasksOpen(false)}
@@ -14,6 +14,13 @@ import { app } from "../lib/bridge";
import { useT } from "../lib/i18n";
import type { TaskEvent, TaskSnapshot } from "../lib/types";
type CatalogTask = TaskSnapshot & { __projectKey: string; __projectLabel: string; __catalogKey: string };
function hasTaskCatalogBinding(): boolean {
const bound = (window as unknown as { go?: { main?: { App?: { ListTaskPage?: unknown } } } }).go?.main?.App?.ListTaskPage;
return typeof bound === "function";
}
// --- helpers ---
type TaskTimerSnapshot = TaskSnapshot & { runtime_lease_until?: string };
@@ -107,6 +114,7 @@ export function TaskMonitorPanel({
onClose,
onOpenSession,
initialOpen = false,
initialScope = "session",
popover = false,
summaryMode = false,
}: {
@@ -114,11 +122,17 @@ export function TaskMonitorPanel({
onClose?: () => void;
onOpenSession?: (tabID: string, taskID: string) => Promise<boolean> | boolean;
initialOpen?: boolean;
initialScope?: "session" | "project" | "all";
popover?: boolean;
summaryMode?: boolean;
}) {
const t = useT();
const [tasks, setTasks] = useState<TaskSnapshot[]>([]);
const [tasks, setTasks] = useState<CatalogTask[]>([]);
const [scope, setScope] = useState<"session" | "project" | "all">(initialScope);
const [query, setQuery] = useState("");
const [nextCursor, setNextCursor] = useState("");
const [indexProgress, setIndexProgress] = useState<{ indexed: number; total: number; partial: boolean }>({ indexed: 0, total: 0, partial: true });
const requestSeq = useRef(0);
const [loading, setLoading] = useState(true);
const [error, setError] = useState<string | null>(null);
const [expanded, setExpanded] = useState<Set<string>>(new Set());
@@ -127,7 +141,7 @@ export function TaskMonitorPanel({
const [actionError, setActionError] = useState<string | null>(null);
const [actionMessage, setActionMessage] = useState<string | null>(null);
const [nowMs, setNowMs] = useState(() => Date.now());
const [pendingAction, setPendingAction] = useState<{ task: TaskSnapshot; action: "stop" | "cancel" } | null>(null);
const [pendingAction, setPendingAction] = useState<{ task: CatalogTask; action: "stop" | "cancel" } | null>(null);
// Per-task event state
const [taskEvents, setTaskEvents] = useState<Map<string, TaskEvent[]>>(
@@ -139,20 +153,36 @@ export function TaskMonitorPanel({
);
const eventCursors = useRef<Map<string, number>>(new Map());
const fetchTasks = useCallback(async () => {
const fetchTasks = useCallback(async (cursor = "") => {
const seq = ++requestSeq.current;
try {
setError(null);
const list = await app.ListTasksForTab(tabID);
setTasks(list ?? []);
if (!hasTaskCatalogBinding()) {
const legacy = await app.ListTasksForTab(tabID);
if (seq !== requestSeq.current) return;
const filtered = legacy.filter((task) => !query.trim() || [task.task_id, task.session_id, task.error_code, task.error_summary].some((value) => (value || "").toLowerCase().includes(query.trim().toLowerCase())));
setTasks(filtered.map((task) => ({ ...task, __projectKey: "", __projectLabel: "", __catalogKey: task.task_id })));
setNextCursor("");
setIndexProgress({ indexed: filtered.length, total: filtered.length, partial: false });
return;
}
const page = await app.ListTaskPage({ scope, tabId: tabID, projectKey: "", states: [], query, cursor, limit: 50 });
if (seq !== requestSeq.current) return;
const decorated = (page.items ?? []).map((item) => ({ ...item.task, __projectKey: item.projectKey, __projectLabel: item.projectLabel, __catalogKey: `${item.projectKey}:${item.task.task_id}` }));
setTasks((current) => cursor ? [...current, ...decorated.filter((item) => !current.some((existing) => existing.__catalogKey === item.__catalogKey))] : decorated);
setNextCursor(page.nextCursor || "");
setIndexProgress({ indexed: page.status.indexed, total: page.status.total, partial: page.partial });
} catch (e) {
if (seq !== requestSeq.current) return;
setError(String(e));
} finally {
setLoading(false);
}
}, [tabID]);
}, [query, scope, tabID]);
// Fetch events for a single task, using afterSequence for incremental load.
const fetchEvents = useCallback(async (taskID: string) => {
const fetchEvents = useCallback(async (task: CatalogTask) => {
const taskID = task.__catalogKey;
setEventsLoading((prev) => new Set(prev).add(taskID));
setEventsError((prev) => {
const next = new Map(prev);
@@ -161,7 +191,9 @@ export function TaskMonitorPanel({
});
try {
const cursor = eventCursors.current.get(taskID) ?? 0;
const events = await app.ListTaskEventsForTab(tabID, taskID, cursor);
const events = hasTaskCatalogBinding()
? (await app.ListTaskEventPage({ projectKey: task.__projectKey, taskId: task.task_id, after: cursor, limit: 50 })).items ?? []
: await app.ListTaskEventsForTab(tabID, task.task_id, cursor);
if (events.length > 0) {
setTaskEvents((prev) => {
const next = new Map(prev);
@@ -193,20 +225,16 @@ export function TaskMonitorPanel({
return next;
});
}
}, [tabID]);
}, [tabID]);
// Initial fetch + periodic polling
useEffect(() => {
fetchTasks();
void fetchTasks("");
const interval = setInterval(() => {
fetchTasks();
// Also refresh events for expanded tasks
expanded.forEach((id) => {
fetchEvents(id);
});
void fetchTasks("");
}, POLL_INTERVAL_MS);
return () => clearInterval(interval);
}, [fetchTasks, fetchEvents, expanded]);
}, [fetchTasks]);
// Live tasks need a ticking clock; terminal and queued tasks stay frozen at
// their persisted end/update time.
@@ -216,7 +244,8 @@ export function TaskMonitorPanel({
return () => clearInterval(interval);
}, [tasks]);
const toggleTask = (id: string) => {
const toggleTask = (task: CatalogTask) => {
const id = task.__catalogKey;
setExpanded((prev) => {
const next = new Set(prev);
if (next.has(id)) {
@@ -225,14 +254,14 @@ export function TaskMonitorPanel({
next.add(id);
// Load events on first expand
if (!taskEvents.has(id)) {
fetchEvents(id);
void fetchEvents(task);
}
}
return next;
});
};
const controlTask = async (task: TaskSnapshot, action: "stop" | "cancel" | "requeue" | "open") => {
const controlTask = async (task: CatalogTask, action: "stop" | "cancel" | "requeue" | "open") => {
if ((action === "stop" || action === "cancel") && (!pendingAction || pendingAction.task.task_id !== task.task_id || pendingAction.action !== action)) {
setPendingAction({ task, action });
return;
@@ -242,18 +271,27 @@ export function TaskMonitorPanel({
setActionError(null);
setActionMessage(null);
try {
if (action === "open" && onOpenSession) {
if (action === "open" && onOpenSession && scope === "session") {
const opened = await onOpenSession(tabID, task.task_id);
if (opened) onClose?.();
return;
}
const result = action === "stop"
? await app.StopTaskForTab(tabID, task.task_id, task.version, "desktop request", `desktop-${action}-${task.task_id}-${task.version}`)
: action === "cancel"
? await app.CancelTaskForTab(tabID, task.task_id, task.version, "desktop request", `desktop-${action}-${task.task_id}-${task.version}`)
: action === "requeue"
? await app.RequeueTaskForTab(tabID, task.task_id, task.version, `desktop-${action}-${task.task_id}-${task.version}`)
: await app.OpenTaskSessionForTab(tabID, task.task_id);
const request = { projectKey: task.__projectKey, taskId: task.task_id, expectedVersion: task.version, reason: "desktop request", idempotencyKey: `desktop-${action}-${task.task_id}-${task.version}` };
const result = hasTaskCatalogBinding()
? action === "stop"
? await app.StopTaskByKey(request)
: action === "cancel"
? await app.CancelTaskByKey(request)
: action === "requeue"
? await app.RequeueTaskByKey(request)
: await app.OpenTaskSessionByKey({ projectKey: task.__projectKey, taskId: task.task_id })
: action === "stop"
? await app.StopTaskForTab(tabID, task.task_id, task.version, request.reason, request.idempotencyKey)
: action === "cancel"
? await app.CancelTaskForTab(tabID, task.task_id, task.version, request.reason, request.idempotencyKey)
: action === "requeue"
? await app.RequeueTaskForTab(tabID, task.task_id, task.version, request.idempotencyKey)
: await app.OpenTaskSessionForTab(tabID, task.task_id);
if (result.error) {
setActionError(`${result.error.code}: ${result.error.message}`);
} else if (action === "open") {
@@ -262,7 +300,7 @@ export function TaskMonitorPanel({
setActionMessage(`Session: ${sessionID}`);
} else {
setActionMessage(result.idempotent ? "Already applied" : "Task updated");
await fetchTasks();
await fetchTasks("");
}
} catch (e) {
setActionError(String(e));
@@ -291,9 +329,9 @@ export function TaskMonitorPanel({
<span className="taskmonitor__count">{tasks.length}</span>
<button
className="taskmonitor__refresh"
onClick={() => {
onClick={() => {
setLoading(true);
fetchTasks();
void fetchTasks("");
}}
title={t("summary.refresh")}
aria-label={t("summary.refresh")}
@@ -314,6 +352,17 @@ export function TaskMonitorPanel({
{open && (
<div className="taskmonitor__body">
{!summaryMode && (
<div className="taskmonitor__filters">
<select value={scope} onChange={(event) => setScope(event.target.value as "session" | "project" | "all")} aria-label="Task scope">
<option value="session">Current session</option>
<option value="project">Current project</option>
<option value="all">All projects</option>
</select>
<input value={query} onChange={(event) => setQuery(event.target.value)} placeholder="Filter tasks" aria-label="Filter tasks" />
</div>
)}
{indexProgress.partial && <div className="taskmonitor__indexing">Indexing tasks ({indexProgress.indexed}/{indexProgress.total})</div>}
{summaryMode && <div className="taskmonitor__category-title">{t("summary.tasks")}</div>}
{actionError && <div className="taskmonitor__state taskmonitor__state--error">{actionError}</div>}
{actionMessage && <div className="taskmonitor__state">{actionMessage}</div>}
@@ -342,21 +391,22 @@ export function TaskMonitorPanel({
sorted.map((task) => {
const cfg = stateConfig(task.state, t);
const runtime = runtimeConfig(task.runtime_state, t);
const isOpen = expanded.has(task.task_id);
const taskKey = task.__catalogKey;
const isOpen = expanded.has(taskKey);
const terminal = isTerminalState(task.state);
const evs = taskEvents.get(task.task_id) ?? [];
const evLoading = eventsLoading.has(task.task_id);
const evError = eventsError.get(task.task_id);
const evs = taskEvents.get(taskKey) ?? [];
const evLoading = eventsLoading.has(taskKey);
const evError = eventsError.get(taskKey);
return (
<div
key={task.task_id}
key={taskKey}
className={`taskmonitor__task taskmonitor__task--${safeStateClass(task.state)}`}
>
<div className="taskmonitor__task-head">
<button
className="taskmonitor__expand"
onClick={() => toggleTask(task.task_id)}
onClick={() => toggleTask(task)}
aria-expanded={isOpen}
aria-label={t("summary.taskLabel", { id: shortID(task.task_id), state: cfg.label })}
>
@@ -369,6 +419,7 @@ export function TaskMonitorPanel({
<span className="taskmonitor__id">
{shortID(task.task_id)}
</span>
{scope === "all" && <span className="taskmonitor__project">{task.__projectLabel}</span>}
<span
className="taskmonitor__badge"
style={{
@@ -438,8 +489,8 @@ export function TaskMonitorPanel({
<span className="taskmonitor__events-count">
{evs.length}
</span>
)}
</div>
)}
</div>
{evLoading && evs.length === 0 && (
<div className="taskmonitor__state">
@@ -509,6 +560,11 @@ export function TaskMonitorPanel({
</div>
);
})}
{nextCursor && !loading && !error && (
<button className="taskmonitor__load-more" onClick={() => void fetchTasks(nextCursor)}>
Load more
</button>
)}
</div>
)}
</div>
+3 -1
View File
@@ -9,6 +9,7 @@ import { addBreadcrumb } from "./breadcrumbs";
import { maybeShare } from "./queryCoalesce";
import { makeMockSessionCatalogBindings } from "./sessionCatalogBridge";
import { makeMockHistoryCatalogBindings, type HistoryCatalogBindings } from "./historyCatalogBridge";
import { makeMockTaskCatalogBindings, type TaskCatalogBindings } from "./taskCatalogBridge";
import { t } from "./i18n";
import { providerIsConfigured, providerRequiresKey, removeProviderAccessesForMock } from "./providerModels";
import { DEFAULT_STATUS_BAR_ITEMS, normalizeStatusBarItems } from "./statusBarItems";
@@ -163,7 +164,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, HistoryCatalogBindings {
export interface AppBindings extends SessionCatalogBindings, HistoryCatalogBindings, TaskCatalogBindings {
Platform(): Promise<string>;
MinimiseMainWindow(): Promise<void>;
ToggleMaximiseMainWindow(): Promise<void>;
@@ -4905,6 +4906,7 @@ function makeMockApp(): AppBindings {
async RequeueTask() { return { schema_version: 1, command: "requeue", task_id: "", accepted: false, idempotent: false, error: { code: "mock", message: "not available in browser mock" } }; },
async OpenTaskSession() { return { schema_version: 1, command: "open_session", task_id: "", accepted: false, idempotent: false, error: { code: "mock", message: "not available in browser mock" } }; },
async ListTasksForTab() { return []; },
...makeMockTaskCatalogBindings(),
async ListTaskEventsForTab() { return []; },
async StopTaskForTab() { return { schema_version: 1, command: "stop", task_id: "", accepted: false, idempotent: false, error: { code: "mock", message: "not available in browser mock" } }; },
async CancelTaskForTab() { return { schema_version: 1, command: "cancel", task_id: "", accepted: false, idempotent: false, error: { code: "mock", message: "not available in browser mock" } }; },
@@ -0,0 +1,27 @@
import type { ControlResult } from "./types";
import type {
TaskActionRequest, TaskCatalogStatus, TaskEventPage, TaskEventPageRequest, TaskPage, TaskPageRequest,
} from "./taskCatalogTypes";
export interface TaskCatalogBindings {
ListTaskPage(req: TaskPageRequest): Promise<TaskPage>;
ListTaskEventPage(req: TaskEventPageRequest): Promise<TaskEventPage>;
StopTaskByKey(req: TaskActionRequest): Promise<ControlResult>;
CancelTaskByKey(req: TaskActionRequest): Promise<ControlResult>;
RequeueTaskByKey(req: TaskActionRequest): Promise<ControlResult>;
OpenTaskSessionByKey(req: { projectKey: string; taskId: string }): Promise<ControlResult>;
GetTaskCatalogStatus(): Promise<TaskCatalogStatus>;
RebuildTaskCatalog(): Promise<void>;
}
export function makeMockTaskCatalogBindings(): TaskCatalogBindings {
const unavailable = (command: string): ControlResult => ({ schema_version: 1, command, task_id: "", accepted: false, idempotent: false, error: { code: "mock", message: "not available in browser mock" } });
return {
async ListTaskPage() { return { items: [], nextCursor: "", revision: 1, partial: false, staleCursor: false, status: await this.GetTaskCatalogStatus() }; },
async ListTaskEventPage(req) { return { items: [], nextSequence: req.after, partial: false }; },
async StopTaskByKey() { return unavailable("stop"); }, async CancelTaskByKey() { return unavailable("cancel"); },
async RequeueTaskByKey() { return unavailable("requeue"); }, async OpenTaskSessionByKey() { return unavailable("open_session"); },
async GetTaskCatalogStatus() { return { state: "ready", mode: "memory", revision: 1, indexed: 0, total: 0, pending: 0, failed: 0 }; },
async RebuildTaskCatalog() {},
};
}
@@ -0,0 +1,14 @@
import type { TaskEvent, TaskSnapshot } from "./types";
export interface TaskPageRequest {
scope: "session" | "project" | "all" | string; tabId: string; projectKey: string; states: string[]; query: string; cursor: string; limit: number;
}
export interface TaskCatalogStatus {
state: string; mode: "disk" | "memory" | string; path?: string; revision: number;
indexed: number; total: number; pending: number; failed: number; lastError?: string;
}
export interface TaskCatalogItem { projectKey: string; projectLabel: string; task: TaskSnapshot }
export interface TaskPage { items: TaskCatalogItem[]; nextCursor: string; revision: number; partial: boolean; staleCursor: boolean; status: TaskCatalogStatus }
export interface TaskEventPageRequest { projectKey: string; taskId: string; after: number; limit: number }
export interface TaskEventPage { items: TaskEvent[]; nextSequence: number; partial: boolean }
export interface TaskActionRequest { projectKey: string; taskId: string; expectedVersion: number; reason: string; idempotencyKey: string }
+2
View File
@@ -2226,6 +2226,8 @@ export interface TaskSnapshot {
error_summary?: string;
}
export type { TaskActionRequest, TaskCatalogItem, TaskCatalogStatus, TaskEventPage, TaskEventPageRequest, TaskPage, TaskPageRequest } from "./taskCatalogTypes";
export interface ControlResult {
schema_version: number;
command: string;
+1
View File
@@ -2381,6 +2381,7 @@ export const en = {
"palette.cmd.models": "Models",
"palette.cmd.reloadRuntime": "Reload Runtime",
"palette.cmd.usageStats": "Usage statistics",
"palette.cmd.taskCenter": "Task Center",
"palette.group.remote": "Remote SSH",
"palette.group.extensions": "Extensions",
"ext.form.defaultTitle": "Extension request",
+1
View File
@@ -3022,6 +3022,7 @@ export const zhTW: Record<DictKey, string> = {
"palette.cmd.models": "模型",
"palette.cmd.reloadRuntime": "重載執行階段",
"palette.cmd.usageStats": "使用統計",
"palette.cmd.taskCenter": "任務中心",
"palette.group.remote": "遠端 SSH",
"palette.group.extensions": "擴充功能",
"ext.form.defaultTitle": "擴充功能請求",
+1
View File
@@ -2384,6 +2384,7 @@ export const zh: Record<DictKey, string> = {
"palette.cmd.models": "模型",
"palette.cmd.reloadRuntime": "重载运行时",
"palette.cmd.usageStats": "使用统计",
"palette.cmd.taskCenter": "任务中心",
"palette.group.remote": "远程 SSH",
"palette.group.extensions": "扩展",
"ext.form.defaultTitle": "扩展请求",
+14 -1
View File
@@ -12,6 +12,7 @@ import (
"reasonix/internal/config"
"reasonix/internal/history"
"reasonix/internal/sessioncatalog"
"reasonix/internal/taskcatalog"
)
type SessionCatalogStatus struct {
@@ -62,7 +63,14 @@ type ProjectTreeChangedV2 struct {
}
func flushDesktopDerivedCatalogs(ctx context.Context) error {
return history.FlushSharedCatalog(ctx)
var first error
if err := history.FlushSharedCatalog(ctx); err != nil && first == nil {
first = err
}
if err := taskcatalog.ShutdownShared(ctx); err != nil && first == nil {
first = err
}
return first
}
func sessionCatalogStatus(status sessioncatalog.Status) SessionCatalogStatus {
@@ -113,6 +121,11 @@ func (a *App) startSessionCatalog(rebuild bool) {
path := sessioncatalog.DefaultPath()
targets := a.sessionCatalogTargets()
history.RegisterCatalogRoots(historyCatalogRoots(targets))
projects := loadProjectsFile()
taskcatalog.RegisterSharedProject(globalWorkspaceRoot(), projects.GlobalTitle)
for _, project := range projects.Projects {
taskcatalog.RegisterSharedProject(project.Root, projectDisplayName(project))
}
if rebuild {
if _, err := sessioncatalog.Rebuild(ctx, path, targets); err != nil && !errors.Is(err, context.Canceled) {
slog.Warn("desktop: rebuild session catalog", "err", err)
+1
View File
@@ -1950,6 +1950,7 @@ func (a *App) buildSettingReplacementController(tab *WorkspaceTab, snap tabRunti
RuntimeReload: boot.RuntimeReload{ForceFullRebuild: reload},
AutoPricingCurrency: a.desktopAutoPricingCurrency(),
StatsSource: "desktop",
TaskStore: a.taskStore(),
Sink: snap.sink,
WorkspaceRoot: snap.workspaceRoot,
SessionDir: sessionDirForSnapshot(snap),
+1
View File
@@ -3904,6 +3904,7 @@ func (a *App) buildTabControllerWithContextCore(tab *WorkspaceTab, loadedSession
RequireKey: false,
AutoPricingCurrency: a.desktopAutoPricingCurrency(),
StatsSource: "desktop",
TaskStore: a.taskStore(),
Sink: sink,
WorkspaceRoot: root,
SessionDir: sessionDir,
+281
View File
@@ -0,0 +1,281 @@
package main
import (
"fmt"
"strings"
"reasonix/internal/agent"
"reasonix/internal/control"
"reasonix/internal/taskcatalog"
"reasonix/internal/taskmonitor"
)
type TaskPageRequest struct {
Scope string `json:"scope"`
TabID string `json:"tabId"`
ProjectKey string `json:"projectKey"`
States []string `json:"states"`
Query string `json:"query"`
Cursor string `json:"cursor"`
Limit int `json:"limit"`
}
type TaskCatalogItem = taskcatalog.Item
type TaskCatalogStatus = taskcatalog.Status
type TaskPage struct {
Items []TaskCatalogItem `json:"items"`
NextCursor string `json:"nextCursor"`
Revision uint64 `json:"revision"`
Partial bool `json:"partial"`
StaleCursor bool `json:"staleCursor"`
Status TaskCatalogStatus `json:"status"`
}
type TaskEventPageRequest struct {
ProjectKey string `json:"projectKey"`
TaskID string `json:"taskId"`
After int `json:"after"`
Limit int `json:"limit"`
}
type TaskEventPage = taskcatalog.EventPage
type TaskActionRequest struct {
ProjectKey string `json:"projectKey"`
TaskID string `json:"taskId"`
ExpectedVersion uint64 `json:"expectedVersion"`
Reason string `json:"reason"`
IdempotencyKey string `json:"idempotencyKey"`
}
type TaskOpenRequest struct {
ProjectKey string `json:"projectKey"`
TaskID string `json:"taskId"`
}
func (a *App) GetTaskCatalogStatus() TaskCatalogStatus {
if catalog := taskcatalog.Shared(); catalog != nil {
return catalog.Status()
}
return TaskCatalogStatus{State: "opening", Mode: "memory", Pending: 1}
}
func (a *App) taskProjectKeys(req TaskPageRequest) ([]string, string, error) {
switch strings.TrimSpace(req.Scope) {
case "session":
target, err := a.taskMonitorTargetForTab(req.TabID)
if err != nil {
return nil, "", err
}
key := taskcatalog.RegisterSharedProject(target.projectDir, workspaceName(target.projectDir))
return []string{key}, target.sessionID, nil
case "project", "":
key := strings.TrimSpace(req.ProjectKey)
if key == "" {
root := a.projectDir()
key = taskcatalog.RegisterSharedProject(root, workspaceName(root))
} else if !a.allowedTaskProjectKey(key) {
return nil, "", fmt.Errorf("unknown project key")
}
return []string{key}, "", nil
case "all":
projects := loadProjectsFile()
keys := []string{taskcatalog.RegisterSharedProject(globalWorkspaceRoot(), projects.GlobalTitle)}
for _, project := range projects.Projects {
keys = append(keys, taskcatalog.RegisterSharedProject(project.Root, projectDisplayName(project)))
}
return keys, "", nil
default:
return nil, "", fmt.Errorf("unknown task scope %q", req.Scope)
}
}
func (a *App) allowedTaskProjectKey(key string) bool {
key = strings.TrimSpace(key)
if key == "" {
return false
}
allowedRoots := []string{globalWorkspaceRoot(), a.projectDir()}
for _, project := range loadProjectsFile().Projects {
allowedRoots = append(allowedRoots, project.Root)
}
for _, root := range allowedRoots {
if taskcatalog.ProjectKey(root) == key {
return true
}
}
return false
}
func (a *App) ListTaskPage(req TaskPageRequest) TaskPage {
status := a.GetTaskCatalogStatus()
out := TaskPage{Items: []TaskCatalogItem{}, Status: status, Partial: true}
catalog := taskcatalog.Shared()
if catalog == nil {
return out
}
keys, sessionID, err := a.taskProjectKeys(req)
if err != nil {
out.Status.LastError = err.Error()
return out
}
for _, key := range keys {
if _, ok, lookupErr := catalog.Project(a.bootContext(), key); lookupErr != nil || !ok {
out.Status.LastError = "unknown project key"
return out
}
}
page, err := catalog.ListPage(a.bootContext(), taskcatalog.PageRequest{ProjectKeys: keys, SessionID: sessionID,
States: req.States, Query: req.Query, Cursor: req.Cursor, Limit: req.Limit})
if err != nil {
out.Status.LastError = err.Error()
return out
}
a.overlayTaskCatalogRuntime(page.Items)
return TaskPage{Items: page.Items, NextCursor: page.NextCursor, Revision: page.Revision,
Partial: page.Partial, StaleCursor: page.StaleCursor, Status: page.Status}
}
func (a *App) overlayTaskCatalogRuntime(items []TaskCatalogItem) {
type controllerRef struct {
projectKey string
sessionID string
ctrl control.SessionAPI
}
a.mu.RLock()
tabs := a.runtimeTabsLocked()
controllers := make([]controllerRef, 0, len(tabs))
for _, tab := range tabs {
if tab == nil || tab.Ctrl == nil {
continue
}
root := strings.TrimSpace(tab.WorkspaceRoot)
if root == "" {
root = globalWorkspaceRoot()
}
controllers = append(controllers, controllerRef{projectKey: taskcatalog.ProjectKey(root), ctrl: tab.Ctrl})
}
a.mu.RUnlock()
for i := range controllers {
controllers[i].sessionID = agent.BranchID(controllers[i].ctrl.SessionPath())
}
type runtimeValue struct{ status string }
running := map[string]runtimeValue{}
for _, controller := range controllers {
for _, job := range controller.ctrl.Jobs() {
key := controller.projectKey + "\x00" + controller.sessionID + "\x00" + job.ID
running[key] = runtimeValue{status: job.Status}
}
}
for i := range items {
key := items[i].ProjectKey + "\x00" + items[i].Task.SessionID + "\x00" + items[i].Task.JobID
value, ok := running[key]
if !ok {
continue
}
items[i].Task.RuntimeState = taskmonitor.RuntimeStateAlive
switch value.status {
case "running":
items[i].Task.State = taskmonitor.TaskStateRunning
case "done":
items[i].Task.RuntimeState = taskmonitor.RuntimeStateExited
items[i].Task.State = taskmonitor.TaskStateSucceeded
case "failed":
items[i].Task.RuntimeState = taskmonitor.RuntimeStateExited
items[i].Task.State = taskmonitor.TaskStateFailed
case "killed", "interrupted":
items[i].Task.RuntimeState = taskmonitor.RuntimeStateExited
items[i].Task.State = taskmonitor.TaskStateCancelled
}
}
}
func (a *App) ListTaskEventPage(req TaskEventPageRequest) TaskEventPage {
out := TaskEventPage{Items: []taskmonitor.TaskEvent{}, NextSequence: req.After}
if !a.allowedTaskProjectKey(req.ProjectKey) {
out.Partial = true
return out
}
catalog := taskcatalog.Shared()
if catalog == nil {
out.Partial = true
return out
}
page, err := catalog.ListEventPage(a.bootContext(), req.ProjectKey, req.TaskID, req.After, req.Limit)
if err != nil {
out.Partial = true
return out
}
return page
}
func (a *App) taskActionProject(key string) (taskcatalog.Project, error) {
if !a.allowedTaskProjectKey(key) {
return taskcatalog.Project{}, fmt.Errorf("unknown project key")
}
catalog := taskcatalog.Shared()
if catalog == nil {
return taskcatalog.Project{}, fmt.Errorf("task catalog is still opening")
}
project, ok, err := catalog.Project(a.bootContext(), strings.TrimSpace(key))
if err != nil {
return project, err
}
if !ok {
return project, fmt.Errorf("unknown project key")
}
return project, nil
}
func (a *App) StopTaskByKey(req TaskActionRequest) (taskmonitor.ControlResult, error) {
project, err := a.taskActionProject(req.ProjectKey)
if err != nil {
return taskmonitor.ControlResult{}, err
}
return a.taskControl().StopTaskWithKiller(a.bootContext(), project.Root, req.TaskID, req.ExpectedVersion, req.Reason, req.IdempotencyKey,
desktopTaskJobKiller{app: a, projectDir: project.Root})
}
func (a *App) CancelTaskByKey(req TaskActionRequest) (taskmonitor.ControlResult, error) {
project, err := a.taskActionProject(req.ProjectKey)
if err != nil {
return taskmonitor.ControlResult{}, err
}
return a.taskControl().CancelTaskWithKiller(a.bootContext(), project.Root, req.TaskID, req.ExpectedVersion, req.Reason, req.IdempotencyKey,
desktopTaskJobKiller{app: a, projectDir: project.Root})
}
func (a *App) RequeueTaskByKey(req TaskActionRequest) (taskmonitor.ControlResult, error) {
project, err := a.taskActionProject(req.ProjectKey)
if err != nil {
return taskmonitor.ControlResult{}, err
}
return a.taskControl().RequeueTask(a.bootContext(), project.Root, req.TaskID, req.ExpectedVersion, req.IdempotencyKey)
}
func (a *App) OpenTaskSessionByKey(req TaskOpenRequest) (taskmonitor.ControlResult, error) {
project, err := a.taskActionProject(req.ProjectKey)
if err != nil {
return taskmonitor.ControlResult{}, err
}
return a.taskControl().OpenTaskSession(a.bootContext(), project.Root, req.TaskID)
}
func (a *App) RebuildTaskCatalog() error {
projects := loadProjectsFile()
items := []struct{ root, label string }{{globalWorkspaceRoot(), projects.GlobalTitle}}
for _, project := range projects.Projects {
items = append(items, struct{ root, label string }{project.Root, projectDisplayName(project)})
}
for _, item := range items {
taskcatalog.RegisterSharedProject(item.root, item.label)
if catalog := taskcatalog.Shared(); catalog != nil {
if err := catalog.RequestReconcileProject(a.bootContext(), item.root, item.label); err != nil {
return err
}
}
}
return nil
}
+10
View File
@@ -480,6 +480,16 @@ Usage statistics use a separate disposable rollup projection:
reasonix catalogs reindex usage [--json]
See [Usage Catalog](./USAGE_CATALOG.md).
Inspect or rebuild the disposable task projection independently:
```sh
reasonix doctor catalogs [--json]
reasonix catalogs reindex tasks [--project PATH ...] [--json]
```
See [Task Catalog](./TASK_CATALOG.md) for the authoritative FileStore boundary,
cross-project routing, and rebuild behavior.
### Memory diagnostics and recovery
Bare `/memory` shows all active project/global facts without hiding same-name
+10
View File
@@ -418,6 +418,16 @@ reasonix catalogs reindex history [--dir PATH ...] [--json]
reasonix catalogs reindex usage [--json]
详见 [用量 Catalog](./USAGE_CATALOG.zh-CN.md)。
可以独立检查或重建可丢弃的 Task 投影:
```sh
reasonix doctor catalogs [--json]
reasonix catalogs reindex tasks [--project PATH ...] [--json]
```
权威 FileStore 边界、跨项目路由和重建行为见
[Task Catalog](./TASK_CATALOG.zh-CN.md)。
### 记忆诊断与恢复
直接运行 `/memory` 会显示全部 project/global active facts,不会隐藏跨 scope 的同名条目。
+3
View File
@@ -43,6 +43,7 @@ non-destructively when `<Reasonix home>/.env` is missing them.
| Disposable session catalog | `<cache root>/session-catalog/v1.sqlite` |
| Disposable history search catalog | `<cache root>/history-search/v1.sqlite` |
| Disposable usage catalog | `<cache root>/usage-catalog/v1.sqlite` |
| Disposable task catalog | `<cache root>/task-catalog/v1.sqlite` |
`<state root>` defaults to `<Reasonix home>`. It only differs when
`REASONIX_STATE_HOME` is set.
@@ -53,6 +54,8 @@ authoritative. See [Session Catalog and Desktop Startup](./SESSION_CATALOG.md).
The history projection is documented in
[History Search Catalog](./HISTORY_SEARCH_CATALOG.md).
The usage rollup projection is documented in [Usage Catalog](./USAGE_CATALOG.md).
Task snapshots and event logs likewise remain authoritative; the rebuildable
cross-project projection is documented in [Task Catalog](./TASK_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
+8
View File
@@ -33,10 +33,18 @@ Legacy 迁移、OS home 约定目录扫描以及其他 fallback 路径都会跳
| 会话 | `<state root>/sessions/` |
| 归档 | `<state root>/archive/` |
| 记忆 | `<state root>/memory/` 与 `<state root>/projects/` |
| 可丢弃的会话 Catalog | `<cache root>/session-catalog/v1.sqlite` |
| 可丢弃的 Task Catalog | `<cache root>/task-catalog/v1.sqlite` |
`<state root>` 默认等于 `<Reasonix home>`;只有设置 `REASONIX_STATE_HOME`
时才会不同。
会话 Catalog 是可重建的查询投影,不是用户数据;JSONL、event log、
metadata sidecar 和 `desktop-projects.json` 仍是权威数据。详见
[Session Catalog and Desktop Startup](./SESSION_CATALOG.zh-CN.md)。Task snapshot 和
event log 也仍是权威数据;可重建的跨项目投影见
[Task Catalog](./TASK_CATALOG.zh-CN.md)。
全局用户配置文件名是 `config.toml`。项目本地配置文件仍叫 `reasonix.toml`。
如果有人说“全局 reasonix.toml”,通常指的是 `<Reasonix home>/config.toml`。
+26
View File
@@ -0,0 +1,26 @@
# Task Catalog
Reasonix stores the cross-project task projection at
`<cache root>/task-catalog/v1.sqlite`. Task snapshots, event JSONL, idempotency
files, file locks, version CAS, and leases remain authoritative. SQLite is never
used to accept stop, cancel, requeue, or open-session commands.
An observed `FileStore` sends non-blocking snapshot and event hints only after
the per-task lock is released. Snapshots are indexed eagerly in bounded batches;
event logs are indexed lazily when a task expands, an event query arrives, or a
new event is appended. Byte-offset and sequence checkpoints resume after an
interruption. Corrupt snapshots and event lines degrade only that task.
The Desktop task center pages across current session, project, or all registered
projects. Project keys are full SHA-256 hashes of canonical saved roots and are
resolved before every action. The control service then re-reads the authoritative
snapshot. Current process jobs and controller state always overlay catalog data,
and expired leases are reconciled at read time without rewriting snapshots.
```sh
reasonix doctor catalogs [--json]
reasonix catalogs reindex tasks [--project PATH ...] [--json]
```
Catalog corruption or partial indexing does not disable task control. Reindexing
replaces only the disposable projection.
+21
View File
@@ -0,0 +1,21 @@
# Task Catalog
Reasonix 将跨项目任务投影保存到 `<cache root>/task-catalog/v1.sqlite`。任务 snapshot、
event JSONL、idempotency 文件、文件锁、version CAS 和 lease 继续作为权威数据。SQLite
绝不参与 stop、cancel、requeue 或 open-session 的接受判断。
Observed `FileStore` 只在 per-task lock 释放后发送非阻塞 snapshot/event 提示。Snapshot
按有界批次优先索引;event log 仅在任务展开、事件查询或追加新事件时惰性索引。byte
offset 和 sequence checkpoint 可在强退后续传;坏 snapshot 或 event line 只降级单个任务。
Desktop Task Center 对当前会话、当前项目或全部已注册项目分页。Project key 是 canonical
已保存 root 的完整 SHA-256,每次 action 都先解析 key,再由 control service 重读权威
snapshot。当前进程的 jobs/controller 状态始终覆盖 catalog;过期 lease 只在读取时对账,
不回写 snapshot。
```sh
reasonix doctor catalogs [--json]
reasonix catalogs reindex tasks [--project PATH ...] [--json]
```
Catalog 损坏或未完成索引不会禁用任务控制;reindex 只替换可丢弃投影。
+3
View File
@@ -60,6 +60,7 @@ import (
"reasonix/internal/sessiontemp"
"reasonix/internal/skill"
"reasonix/internal/stats"
"reasonix/internal/taskmonitor"
"reasonix/internal/tool"
"reasonix/internal/tool/builtin"
"reasonix/internal/tool/sessiontool"
@@ -128,6 +129,7 @@ type Options struct {
// StatsSource labels this frontend's usage records (desktop/cli/serve).
// Empty disables usage recording for this controller.
StatsSource string
TaskStore taskmonitor.WriteStore // Authoritative store, never a SQLite catalog.
// ExtraPlugins are session-scoped MCP servers supplied by a host transport
// (for example ACP session/new). They are connected eagerly for this
// controller but are not persisted to reasonix.toml.
@@ -1935,6 +1937,7 @@ func build(ctx context.Context, opts Options) (*BuildResult, error) {
BalanceKey: entry.APIKey(),
BalanceClient: balanceClient,
Jobs: jm,
TaskStore: opts.TaskStore,
WorkspaceLease: workspaceLease,
Registry: reg,
PluginCtx: ctx,
+56
View File
@@ -0,0 +1,56 @@
package cli
import (
"context"
"flag"
"fmt"
"os"
"path/filepath"
"strings"
"reasonix/internal/taskcatalog"
)
func init() {
registerCatalogCommand(catalogCommand{
name: "tasks", path: taskcatalog.DefaultPath, reindex: reindexTaskCatalog,
completionFlags: []cliCompletionFlag{
completionFlag("--json", cliCompletionNoValue),
completionFlag("--project", cliCompletionPathValue),
},
})
}
func reindexTaskCatalog(args []string) int {
fs := flag.NewFlagSet("catalogs reindex tasks", flag.ContinueOnError)
var projects stringListFlag
jsonOut := fs.Bool("json", false, "print status as JSON")
fs.Var(&projects, "project", "project root to index; repeat for multiple projects")
if code, ok := parseCommandFlags(fs, args); !ok {
return code
}
if len(projects) == 0 {
for _, target := range defaultSessionCatalogTargets() {
if strings.TrimSpace(target.WorkspaceRoot) != "" {
projects = append(projects, target.WorkspaceRoot)
}
}
}
catalog, err := taskcatalog.Open(context.Background(), "")
if err != nil {
fmt.Fprintln(os.Stderr, "error:", err)
return 1
}
defer catalog.Close(context.Background())
for _, root := range projects {
project, err := catalog.RegisterProject(context.Background(), root, filepath.Base(root))
if err == nil {
err = catalog.ReconcileProject(context.Background(), project)
}
if err != nil {
fmt.Fprintln(os.Stderr, "error:", err)
return 1
}
}
return printCatalogStatus(catalog.Status(), *jsonOut)
}
+9 -1
View File
@@ -461,6 +461,10 @@ type Options struct {
BalanceClient *http.Client
// Jobs is the session-scoped background-job manager (nil disables background jobs).
Jobs *jobs.Manager
// TaskStore remains a FileStore-compatible authority. Desktop injects one
// observed instance so recorder and task-control APIs share post-commit
// projection hints; nil preserves the ordinary FileStore.
TaskStore taskmonitor.WriteStore
// WorkspaceLease is the Delivery writer owner shared with the executor.
WorkspaceLease *workspacelease.Owner
// Registry is the executor's live tool set, and PluginCtx the session-scoped
@@ -666,8 +670,12 @@ func New(opts Options) *Controller {
// must never affect the agent pipeline. The session id is resolved lazily
// because the session path is only fixed once the first turn begins.
if c.jobs != nil && c.workspaceRoot != "" {
taskStore := opts.TaskStore
if taskStore == nil {
taskStore = taskmonitor.NewFileStore(filepath.Join(".reasonix", "tasks"))
}
c.jobs.SetTaskRecorder(taskmonitor.NewTaskRecorder(
taskmonitor.NewFileStore(filepath.Join(".reasonix", "tasks")),
taskStore,
c.workspaceRoot,
func() string { return c.parentSessionID() },
))
+50
View File
@@ -0,0 +1,50 @@
package taskcatalog
import (
"context"
"encoding/json"
"fmt"
"path/filepath"
"testing"
"time"
"reasonix/internal/taskmonitor"
)
func BenchmarkWarmFirstPageHundredThousandTasks(b *testing.B) {
catalog, err := Open(context.Background(), filepath.Join(b.TempDir(), "tasks.sqlite"))
if err != nil {
b.Fatal(err)
}
b.Cleanup(func() { _ = catalog.Close(context.Background()) })
root := b.TempDir()
project := Project{Key: ProjectKey(root), Root: root, Label: "Benchmark"}
if _, err := catalog.db.Exec(`INSERT INTO task_projects(project_key,project_root,project_label,state) VALUES(?,?,?,'ready')`,
project.Key, project.Root, project.Label); err != nil {
b.Fatal(err)
}
tx, err := catalog.db.Begin()
if err != nil {
b.Fatal(err)
}
now := time.Now().UTC()
for i := range 100_000 {
task := taskmonitor.TaskSnapshot{SchemaVersion: 1, TaskID: fmt.Sprintf("task-%06d", i), SessionID: "session",
State: taskmonitor.TaskStateSucceeded, Version: 1, CreatedAt: now, UpdatedAt: now.Add(time.Duration(i) * time.Millisecond)}
raw, _ := json.Marshal(task)
if _, err := tx.Exec(`INSERT INTO task_snapshots(project_key,task_id,session_id,state,version,created_at,updated_at,snapshot_json)
VALUES(?,?,?,?,?,?,?,?)`, project.Key, task.TaskID, task.SessionID, task.State, task.Version,
task.CreatedAt.UnixMilli(), task.UpdatedAt.UnixMilli(), raw); err != nil {
b.Fatal(err)
}
}
if err := tx.Commit(); err != nil {
b.Fatal(err)
}
b.ResetTimer()
for range b.N {
if _, err := catalog.ListPage(context.Background(), PageRequest{ProjectKeys: []string{project.Key}, Limit: 50}); err != nil {
b.Fatal(err)
}
}
}
+677
View File
@@ -0,0 +1,677 @@
// Package taskcatalog maintains a disposable cross-project projection of the
// authoritative taskmonitor FileStore.
package taskcatalog
import (
"context"
"crypto/sha256"
"database/sql"
"encoding/base64"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"strings"
"sync"
"sync/atomic"
"time"
"reasonix/internal/config"
"reasonix/internal/projectiondb"
"reasonix/internal/taskmonitor"
)
const (
SchemaVersion = 1
DefaultLimit = 50
MaxLimit = 200
missingGrace = 30 * time.Second
)
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"`
}
type Project struct {
Key string `json:"projectKey"`
Root string `json:"projectRoot"`
Label string `json:"projectLabel"`
}
type PageRequest struct {
ProjectKeys []string
SessionID string
States []string
Query string
Cursor string
Limit int
}
type Item struct {
ProjectKey string `json:"projectKey"`
ProjectLabel string `json:"projectLabel"`
Task taskmonitor.TaskSnapshot `json:"task"`
}
type Page struct {
Items []Item `json:"items"`
NextCursor string `json:"nextCursor"`
Revision uint64 `json:"revision"`
Partial bool `json:"partial"`
StaleCursor bool `json:"staleCursor"`
Status Status `json:"status"`
}
type EventPage struct {
Items []taskmonitor.TaskEvent `json:"items"`
NextSequence int `json:"nextSequence"`
Partial bool `json:"partial"`
}
type cursor struct {
Revision uint64 `json:"r"`
Updated int64 `json:"u"`
Project string `json:"p"`
Task string `json:"t"`
}
type request struct {
projectRoot string
taskID string
events bool
flush chan struct{}
}
type Catalog struct {
db *sql.DB
store *taskmonitor.FileStore
ctx context.Context
cancel context.CancelFunc
queue chan request
dirtyWake chan struct{}
dirtyProjects sync.Map
wg sync.WaitGroup
closing atomic.Bool
revision atomic.Uint64
statusMu sync.RWMutex
status Status
reconcileMu sync.Mutex
reconciling map[string]bool
registered map[string]bool
reconcileDone bool
}
func DefaultPath() string { return filepath.Join(config.CacheDir(), "task-catalog", "v1.sqlite") }
func ProjectKey(root string) string {
root = filepath.Clean(strings.TrimSpace(root))
if abs, err := filepath.Abs(root); err == nil {
root = abs
}
sum := sha256.Sum256([]byte(root))
return hex.EncodeToString(sum[:])
}
const schema = `
CREATE TABLE task_state(id INTEGER PRIMARY KEY CHECK(id=1),revision INTEGER NOT NULL DEFAULT 0);
INSERT INTO task_state(id,revision) VALUES(1,0);
CREATE TABLE task_projects(project_key TEXT PRIMARY KEY,project_root TEXT UNIQUE NOT NULL,project_label 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 task_snapshots(project_key TEXT NOT NULL,task_id TEXT NOT NULL,session_id TEXT NOT NULL DEFAULT '',job_id TEXT NOT NULL DEFAULT '',
kind TEXT NOT NULL DEFAULT '',label TEXT NOT NULL DEFAULT '',state TEXT NOT NULL,runtime_state TEXT NOT NULL DEFAULT '',runtime_lease_until INTEGER NOT NULL DEFAULT 0,
version INTEGER NOT NULL,created_at INTEGER NOT NULL,updated_at INTEGER NOT NULL,error_code TEXT NOT NULL DEFAULT '',snapshot_fingerprint TEXT NOT NULL DEFAULT '',
snapshot_json BLOB NOT NULL,health TEXT NOT NULL DEFAULT 'ok',missing_since INTEGER NOT NULL DEFAULT 0,seen_generation INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY(project_key,task_id),FOREIGN KEY(project_key) REFERENCES task_projects(project_key) ON DELETE CASCADE);
CREATE TABLE task_event_sources(project_key TEXT NOT NULL,task_id TEXT NOT NULL,path TEXT NOT NULL,size INTEGER NOT NULL DEFAULT 0,indexed_offset INTEGER NOT NULL DEFAULT 0,
fingerprint TEXT NOT NULL DEFAULT '',state TEXT NOT NULL DEFAULT 'pending',error TEXT NOT NULL DEFAULT '',PRIMARY KEY(project_key,task_id));
CREATE TABLE task_events(project_key TEXT NOT NULL,task_id TEXT NOT NULL,sequence INTEGER NOT NULL,timestamp INTEGER NOT NULL,event_type TEXT NOT NULL,
state TEXT NOT NULL,runtime_state TEXT NOT NULL DEFAULT '',event_json BLOB NOT NULL,PRIMARY KEY(project_key,task_id,sequence));
CREATE INDEX idx_task_project_page ON task_snapshots(project_key,updated_at DESC,task_id);
CREATE INDEX idx_task_state_page ON task_snapshots(state,updated_at DESC,project_key,task_id);
CREATE INDEX idx_task_session ON task_snapshots(session_id,updated_at DESC,task_id);
CREATE INDEX idx_task_events_page ON task_events(project_key,task_id,sequence);
`
func migrations() []projectiondb.Migration {
return []projectiondb.Migration{{Version: 1, Apply: func(ctx context.Context, tx *sql.Tx) error {
_, err := tx.ExecContext(ctx, schema)
return err
}}}
}
func Open(ctx context.Context, path string) (*Catalog, error) {
if path == "" {
path = DefaultPath()
}
handle, err := projectiondb.Open(ctx, projectiondb.OpenOptions{Path: path, MemoryName: "task-catalog", Migrations: migrations(), MaxOpenConns: 4})
if err != nil {
return nil, err
}
workerCtx, cancel := context.WithCancel(context.Background())
c := &Catalog{db: handle.DB, store: taskmonitor.NewFileStore(filepath.Join(".reasonix", "tasks")), ctx: workerCtx, cancel: cancel,
queue: make(chan request, 1024), dirtyWake: make(chan struct{}, 1), reconciling: map[string]bool{}, registered: map[string]bool{}, status: Status{State: string(handle.Status.State), Mode: handle.Status.Mode, Path: handle.Status.Path, LastError: handle.Status.LastError}}
var revision uint64
_ = c.db.QueryRowContext(ctx, `SELECT revision FROM task_state WHERE id=1`).Scan(&revision)
c.revision.Store(revision)
c.refresh(ctx)
c.wg.Add(1)
go c.worker()
return c, nil
}
func (c *Catalog) ObservedStore() *taskmonitor.FileStore {
return taskmonitor.NewObservedFileStore(filepath.Join(".reasonix", "tasks"), c)
}
func (c *Catalog) SnapshotChanged(projectRoot, taskID string) {
c.enqueue(request{projectRoot: projectRoot, taskID: taskID})
}
func (c *Catalog) EventsChanged(projectRoot, taskID string) {
c.enqueue(request{projectRoot: projectRoot, taskID: taskID, events: true})
}
func (c *Catalog) enqueue(req request) {
if c.closing.Load() {
return
}
select {
case c.queue <- req:
default:
c.dirtyProjects.Store(req.projectRoot, true)
c.wakeDirty()
}
}
func (c *Catalog) worker() {
defer c.wg.Done()
ticker := time.NewTicker(time.Minute)
defer ticker.Stop()
for {
if root, ok := c.takeDirtyProject(); ok {
if project, exists, err := c.projectByRoot(c.ctx, root); err == nil && exists {
_ = c.ReconcileProject(c.ctx, project)
} else if err == nil {
_, _ = c.RegisterProject(c.ctx, root, filepath.Base(root))
}
continue
}
select {
case <-c.ctx.Done():
return
case <-ticker.C:
c.markRegisteredProjectsDirty()
case <-c.dirtyWake:
case req := <-c.queue:
if req.flush != nil {
close(req.flush)
continue
}
if req.events {
_ = c.indexEvents(c.ctx, req.projectRoot, req.taskID)
} else {
_ = c.indexSnapshot(c.ctx, req.projectRoot, req.taskID, 0)
}
}
}
}
func (c *Catalog) takeDirtyProject() (string, bool) {
var root string
c.dirtyProjects.Range(func(key, _ any) bool {
root, _ = key.(string)
c.dirtyProjects.Delete(key)
return false
})
return root, root != ""
}
func (c *Catalog) markRegisteredProjectsDirty() {
rows, err := c.db.QueryContext(c.ctx, `SELECT project_root FROM task_projects`)
if err != nil {
return
}
defer rows.Close()
for rows.Next() {
var root string
if rows.Scan(&root) == nil {
c.dirtyProjects.Store(root, true)
}
}
}
func (c *Catalog) wakeDirty() {
select {
case c.dirtyWake <- struct{}{}:
default:
}
}
func (c *Catalog) Flush(ctx context.Context) error {
done := make(chan struct{})
select {
case c.queue <- request{flush: done}:
case <-ctx.Done():
return ctx.Err()
}
select {
case <-done:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
func (c *Catalog) RegisterProject(ctx context.Context, root, label string) (Project, error) {
root = filepath.Clean(strings.TrimSpace(root))
if abs, err := filepath.Abs(root); err == nil {
root = abs
}
project := Project{Key: ProjectKey(root), Root: root, Label: strings.TrimSpace(label)}
_, err := c.db.ExecContext(ctx, `INSERT INTO task_projects(project_key,project_root,project_label,state) VALUES(?,?,?,'pending')
ON CONFLICT(project_key) DO UPDATE SET project_root=excluded.project_root,project_label=excluded.project_label`, project.Key, project.Root, project.Label)
c.reconcileMu.Lock()
firstRegistration := !c.registered[project.Key]
c.registered[project.Key] = true
c.reconcileMu.Unlock()
if err == nil && firstRegistration {
c.scheduleReconcile(project)
}
return project, err
}
// RequestReconcileProject forces a background authority scan even when the
// project was already registered by an earlier page request.
func (c *Catalog) RequestReconcileProject(ctx context.Context, root, label string) error {
project, err := c.RegisterProject(ctx, root, label)
if err != nil {
return err
}
c.scheduleReconcile(project)
return nil
}
func (c *Catalog) scheduleReconcile(project Project) {
c.reconcileMu.Lock()
if c.reconcileDone || c.reconciling[project.Key] {
c.reconcileMu.Unlock()
return
}
c.reconciling[project.Key] = true
c.wg.Add(1)
c.reconcileMu.Unlock()
go func() {
defer func() {
c.reconcileMu.Lock()
delete(c.reconciling, project.Key)
c.reconcileMu.Unlock()
c.wg.Done()
}()
_ = c.ReconcileProject(c.ctx, project)
}()
}
func (c *Catalog) ReconcileProject(ctx context.Context, project Project) error {
tasks, err := c.store.ListTasks(ctx, project.Root)
if err != nil {
return err
}
var generation int64
err = c.db.QueryRowContext(ctx, `UPDATE task_projects SET scan_generation=scan_generation+1,state='scanning',total=? WHERE project_key=? RETURNING scan_generation`, len(tasks), project.Key).Scan(&generation)
if err != nil {
return err
}
for _, task := range tasks {
if err := c.upsertSnapshot(ctx, project, task, generation); err != nil {
continue
}
}
now := time.Now().UnixMilli()
_, _ = c.db.ExecContext(ctx, `UPDATE task_snapshots SET missing_since=CASE WHEN missing_since=0 THEN ? ELSE missing_since END,health='missing'
WHERE project_key=? AND seen_generation<>?`, now, project.Key, generation)
cutoff := now - missingGrace.Milliseconds()
_, _ = c.db.ExecContext(ctx, `DELETE FROM task_events WHERE project_key=? AND task_id IN (
SELECT task_id FROM task_snapshots WHERE project_key=? AND seen_generation<>? AND missing_since>0 AND missing_since<=?
)`, project.Key, project.Key, generation, cutoff)
_, _ = c.db.ExecContext(ctx, `DELETE FROM task_event_sources WHERE project_key=? AND task_id IN (
SELECT task_id FROM task_snapshots WHERE project_key=? AND seen_generation<>? AND missing_since>0 AND missing_since<=?
)`, project.Key, project.Key, generation, cutoff)
_, _ = c.db.ExecContext(ctx, `DELETE FROM task_snapshots WHERE project_key=? AND seen_generation<>? AND missing_since>0 AND missing_since<=?`,
project.Key, generation, cutoff)
tx, beginErr := c.db.BeginTx(ctx, nil)
if beginErr != nil {
return beginErr
}
if _, err = tx.ExecContext(ctx, `UPDATE task_projects SET state='ready',indexed=?,completed_at=? WHERE project_key=?`, len(tasks), now, project.Key); 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.revision.Store(revision)
c.refresh(context.Background())
return nil
}
func (c *Catalog) indexSnapshot(ctx context.Context, root, taskID string, generation int64) error {
project, ok, err := c.projectByRoot(ctx, root)
if err != nil {
return err
}
if !ok {
project, err = c.RegisterProject(ctx, root, filepath.Base(root))
if err != nil {
return err
}
}
task, err := c.store.GetTask(ctx, project.Root, taskID)
if err != nil || task == nil {
return err
}
return c.upsertSnapshot(ctx, project, *task, generation)
}
func (c *Catalog) upsertSnapshot(ctx context.Context, project Project, task taskmonitor.TaskSnapshot, generation int64) error {
b, err := json.Marshal(task)
if err != nil {
return err
}
hash := sha256.Sum256(b)
lease := int64(0)
if !task.RuntimeLeaseUntil.IsZero() {
lease = task.RuntimeLeaseUntil.UnixMilli()
}
tx, err := c.db.BeginTx(ctx, nil)
if err != nil {
return err
}
_, err = tx.ExecContext(ctx, `INSERT INTO task_snapshots(project_key,task_id,session_id,job_id,state,runtime_state,runtime_lease_until,version,
created_at,updated_at,error_code,snapshot_fingerprint,snapshot_json,health,missing_since,seen_generation) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,'ok',0,?)
ON CONFLICT(project_key,task_id) DO UPDATE SET session_id=excluded.session_id,job_id=excluded.job_id,state=excluded.state,
runtime_state=excluded.runtime_state,runtime_lease_until=excluded.runtime_lease_until,version=excluded.version,created_at=excluded.created_at,
updated_at=excluded.updated_at,error_code=excluded.error_code,snapshot_fingerprint=excluded.snapshot_fingerprint,snapshot_json=excluded.snapshot_json,
health='ok',missing_since=0,seen_generation=excluded.seen_generation`, project.Key, task.TaskID, task.SessionID, task.JobID, task.State,
task.RuntimeState, lease, task.Version, task.CreatedAt.UnixMilli(), task.UpdatedAt.UnixMilli(), task.ErrorCode, hex.EncodeToString(hash[:]), b, generation)
if 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.revision.Store(revision)
c.refresh(context.Background())
return nil
}
func (c *Catalog) indexEvents(ctx context.Context, root, taskID string) error {
if taskID == "" || filepath.Base(taskID) != taskID || strings.ContainsAny(taskID, `/\\`) {
return errors.New("invalid task id")
}
project, ok, err := c.projectByRoot(ctx, root)
if err != nil || !ok {
return err
}
path := filepath.Join(project.Root, ".reasonix", "tasks", taskID, "events.jsonl")
info, statErr := os.Stat(path)
if errors.Is(statErr, os.ErrNotExist) {
return nil
}
if statErr != nil {
return statErr
}
fingerprint := fmt.Sprintf("%d:%d", info.Size(), info.ModTime().UnixNano())
var oldPath, oldFingerprint string
var oldSize, offset int64
err = c.db.QueryRowContext(ctx, `SELECT path,size,indexed_offset,fingerprint FROM task_event_sources WHERE project_key=? AND task_id=?`,
project.Key, taskID).Scan(&oldPath, &oldSize, &offset, &oldFingerprint)
reset := errors.Is(err, sql.ErrNoRows) || oldPath != path || info.Size() < oldSize || info.Size() == oldSize && fingerprint != oldFingerprint
if err != nil && !errors.Is(err, sql.ErrNoRows) {
return err
}
if !reset && info.Size() == oldSize && fingerprint == oldFingerprint {
return nil
}
if reset {
offset = 0
}
tail, err := c.store.ReadEventTail(ctx, project.Root, taskID, offset)
if err != nil {
return err
}
reset = reset || tail.Reset
tx, err := c.db.BeginTx(ctx, nil)
if err != nil {
return err
}
if reset {
if _, err := tx.ExecContext(ctx, `DELETE FROM task_events WHERE project_key=? AND task_id=?`, project.Key, taskID); err != nil {
_ = tx.Rollback()
return err
}
}
for _, event := range tail.Items {
b, _ := json.Marshal(event)
if _, err := tx.ExecContext(ctx, `INSERT OR REPLACE INTO task_events(project_key,task_id,sequence,timestamp,event_type,state,runtime_state,event_json)
VALUES(?,?,?,?,?,?,?,?)`, project.Key, taskID, event.Sequence, event.Timestamp.UnixMilli(), event.EventType, event.State, event.RuntimeState, b); err != nil {
_ = tx.Rollback()
return err
}
}
_, err = tx.ExecContext(ctx, `INSERT INTO task_event_sources(project_key,task_id,path,size,indexed_offset,fingerprint,state,error)
VALUES(?,?,?,?,?,?,'ready','') ON CONFLICT(project_key,task_id) DO UPDATE SET path=excluded.path,size=excluded.size,
indexed_offset=excluded.indexed_offset,fingerprint=excluded.fingerprint,state='ready',error=''`, project.Key, taskID, path,
info.Size(), tail.NextOffset, fingerprint)
if err != nil {
_ = tx.Rollback()
return err
}
return tx.Commit()
}
func (c *Catalog) ListPage(ctx context.Context, req PageRequest) (Page, error) {
status := c.Status()
out := Page{Items: []Item{}, Revision: c.revision.Load(), Partial: status.Pending > 0, Status: status}
limit := req.Limit
if limit <= 0 {
limit = DefaultLimit
}
if limit > MaxLimit {
limit = MaxLimit
}
cur, err := decodeCursor(req.Cursor)
if err != nil {
return out, err
}
if cur != nil && cur.Revision != out.Revision {
out.StaleCursor = true
return out, nil
}
where := []string{`s.health='ok'`, `s.missing_since=0`}
args := []any{}
if len(req.ProjectKeys) > 0 {
parts := make([]string, len(req.ProjectKeys))
for i, key := range req.ProjectKeys {
parts[i] = "?"
args = append(args, key)
}
where = append(where, `s.project_key IN (`+strings.Join(parts, ",")+`)`)
}
if req.SessionID != "" {
where = append(where, `s.session_id=?`)
args = append(args, req.SessionID)
}
if len(req.States) > 0 {
parts := make([]string, len(req.States))
for i, state := range req.States {
parts[i] = "?"
args = append(args, state)
}
where = append(where, `s.state IN (`+strings.Join(parts, ",")+`)`)
}
if query := strings.ToLower(strings.TrimSpace(req.Query)); query != "" {
where = append(where, `(lower(s.task_id) LIKE ? OR lower(s.session_id) LIKE ? OR lower(s.error_code) LIKE ?)`)
like := "%" + query + "%"
args = append(args, like, like, like)
}
if cur != nil {
where = append(where, `(s.updated_at<? OR (s.updated_at=? AND s.project_key>?) OR (s.updated_at=? AND s.project_key=? AND s.task_id>?))`)
args = append(args, cur.Updated, cur.Updated, cur.Project, cur.Updated, cur.Project, cur.Task)
}
args = append(args, limit+1)
rows, err := c.db.QueryContext(ctx, `SELECT s.project_key,p.project_label,s.snapshot_json FROM task_snapshots s JOIN task_projects p ON p.project_key=s.project_key
WHERE `+strings.Join(where, ` AND `)+` ORDER BY s.updated_at DESC,s.project_key,s.task_id LIMIT ?`, args...)
if err != nil {
return out, err
}
defer rows.Close()
for rows.Next() {
var item Item
var raw []byte
if err := rows.Scan(&item.ProjectKey, &item.ProjectLabel, &raw); err != nil {
return out, err
}
if json.Unmarshal(raw, &item.Task) != nil {
continue
}
item.Task.ReconcileRuntime(time.Now())
out.Items = append(out.Items, item)
}
if len(out.Items) > limit {
out.Items = out.Items[:limit]
last := out.Items[len(out.Items)-1]
out.NextCursor = encodeCursor(cursor{Revision: out.Revision, Updated: last.Task.UpdatedAt.UnixMilli(), Project: last.ProjectKey, Task: last.Task.TaskID})
}
return out, rows.Err()
}
func (c *Catalog) ListEventPage(ctx context.Context, projectKey, taskID string, after, limit int) (EventPage, error) {
out := EventPage{Items: []taskmonitor.TaskEvent{}, NextSequence: after}
project, ok, err := c.Project(ctx, projectKey)
if err != nil || !ok {
return out, err
}
if err := c.indexEvents(ctx, project.Root, taskID); err != nil {
out.Partial = true
return out, nil
}
if limit <= 0 {
limit = DefaultLimit
}
if limit > MaxLimit {
limit = MaxLimit
}
rows, err := c.db.QueryContext(ctx, `SELECT event_json FROM task_events WHERE project_key=? AND task_id=? AND sequence>? ORDER BY sequence LIMIT ?`, projectKey, taskID, after, limit)
if err != nil {
return out, err
}
defer rows.Close()
for rows.Next() {
var raw []byte
var event taskmonitor.TaskEvent
if err := rows.Scan(&raw); err != nil {
return out, err
}
if json.Unmarshal(raw, &event) == nil {
out.Items = append(out.Items, event)
out.NextSequence = event.Sequence
}
}
return out, rows.Err()
}
func (c *Catalog) Project(ctx context.Context, key string) (Project, bool, error) {
var project Project
err := c.db.QueryRowContext(ctx, `SELECT project_key,project_root,project_label FROM task_projects WHERE project_key=?`, key).Scan(&project.Key, &project.Root, &project.Label)
if errors.Is(err, sql.ErrNoRows) {
return project, false, nil
}
return project, err == nil, err
}
func (c *Catalog) projectByRoot(ctx context.Context, root string) (Project, bool, error) {
return c.Project(ctx, ProjectKey(root))
}
func bump(ctx context.Context, tx *sql.Tx) (uint64, error) {
if _, err := tx.ExecContext(ctx, `UPDATE task_state SET revision=revision+1 WHERE id=1`); err != nil {
return 0, err
}
var revision uint64
err := tx.QueryRowContext(ctx, `SELECT revision FROM task_state WHERE id=1`).Scan(&revision)
return revision, err
}
func encodeCursor(value cursor) string {
b, _ := json.Marshal(value)
return base64.RawURLEncoding.EncodeToString(b)
}
func decodeCursor(value string) (*cursor, error) {
if strings.TrimSpace(value) == "" {
return nil, nil
}
b, err := base64.RawURLEncoding.DecodeString(value)
if err != nil {
return nil, fmt.Errorf("invalid task cursor: %w", err)
}
var out cursor
if json.Unmarshal(b, &out) != nil || out.Task == "" {
return nil, errors.New("invalid task cursor")
}
return &out, nil
}
func (c *Catalog) refresh(ctx context.Context) {
var indexed, total, pending, failed int64
_ = c.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM task_snapshots WHERE health='ok'`).Scan(&indexed)
_ = c.db.QueryRowContext(ctx, `SELECT COALESCE(SUM(total),0) FROM task_projects`).Scan(&total)
_ = c.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM task_projects WHERE state<>'ready'`).Scan(&pending)
_ = c.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM task_snapshots WHERE health='corrupt'`).Scan(&failed)
c.statusMu.Lock()
c.status.Revision, c.status.Indexed, c.status.Total, c.status.Pending, c.status.Failed = c.revision.Load(), indexed, total, pending, failed
c.statusMu.Unlock()
}
func (c *Catalog) Status() Status {
c.statusMu.RLock()
defer c.statusMu.RUnlock()
return c.status
}
func (c *Catalog) Close(ctx context.Context) error {
c.closing.Store(true)
c.cancel()
c.reconcileMu.Lock()
c.reconcileDone = true
c.reconcileMu.Unlock()
done := make(chan struct{})
go func() { c.wg.Wait(); close(done) }()
select {
case <-done:
return c.db.Close()
case <-ctx.Done():
return ctx.Err()
}
}
+97
View File
@@ -0,0 +1,97 @@
package taskcatalog
import (
"context"
"path/filepath"
"testing"
"time"
"reasonix/internal/taskmonitor"
)
func snapshot(id, session string, version uint64, updated time.Time) taskmonitor.TaskSnapshot {
return taskmonitor.TaskSnapshot{SchemaVersion: 1, TaskID: id, SessionID: session, State: taskmonitor.TaskStateRunning,
RuntimeState: taskmonitor.RuntimeStateAlive, RuntimeLeaseUntil: updated.Add(time.Hour), Version: version,
CreatedAt: updated.Add(-time.Minute), UpdatedAt: updated}
}
func TestObservedStoreIndexesSnapshotsAndEvents(t *testing.T) {
t.Parallel()
ctx := context.Background()
projectRoot := t.TempDir()
catalog, err := Open(ctx, filepath.Join(t.TempDir(), "tasks.sqlite"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = catalog.Close(context.Background()) })
project, err := catalog.RegisterProject(ctx, projectRoot, "Demo")
if err != nil {
t.Fatal(err)
}
store := catalog.ObservedStore()
now := time.Now()
if err := store.SaveTask(ctx, projectRoot, snapshot("task-1", "session-1", 1, now)); err != nil {
t.Fatal(err)
}
if err := store.AppendAuditEvent(ctx, projectRoot, taskmonitor.TaskEvent{Timestamp: now, EventType: "state_change", TaskID: "task-1", SessionID: "session-1", State: taskmonitor.TaskStateRunning}); err != nil {
t.Fatal(err)
}
flushCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
defer cancel()
if err := catalog.Flush(flushCtx); err != nil {
t.Fatal(err)
}
page, err := catalog.ListPage(ctx, PageRequest{ProjectKeys: []string{project.Key}, SessionID: "session-1", Limit: 50})
if err != nil || len(page.Items) != 1 || page.Items[0].Task.TaskID != "task-1" {
t.Fatalf("page=%#v err=%v", page, err)
}
events, err := catalog.ListEventPage(ctx, project.Key, "task-1", 0, 50)
if err != nil || len(events.Items) != 1 || events.NextSequence != 1 {
t.Fatalf("events=%#v err=%v", events, err)
}
if err := store.AppendAuditEvent(ctx, projectRoot, taskmonitor.TaskEvent{Timestamp: now.Add(time.Second), EventType: "state_change",
TaskID: "task-1", SessionID: "session-1", State: taskmonitor.TaskStateSucceeded}); err != nil {
t.Fatal(err)
}
if err := catalog.Flush(flushCtx); err != nil {
t.Fatal(err)
}
events, err = catalog.ListEventPage(ctx, project.Key, "task-1", 1, 50)
if err != nil || len(events.Items) != 1 || events.Items[0].Sequence != 2 {
t.Fatalf("incremental events=%#v err=%v", events, err)
}
}
func TestPageCursorIsRevisionBound(t *testing.T) {
t.Parallel()
ctx := context.Background()
root := t.TempDir()
catalog, err := Open(ctx, filepath.Join(t.TempDir(), "tasks.sqlite"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = catalog.Close(context.Background()) })
project, _ := catalog.RegisterProject(ctx, root, "Demo")
store := catalog.ObservedStore()
now := time.Now()
for i, id := range []string{"a", "b", "c"} {
if err := store.SaveTask(ctx, root, snapshot(id, "session", 1, now.Add(time.Duration(i)*time.Minute))); err != nil {
t.Fatal(err)
}
}
flushCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
defer cancel()
_ = catalog.Flush(flushCtx)
first, err := catalog.ListPage(ctx, PageRequest{ProjectKeys: []string{project.Key}, Limit: 2})
if err != nil || len(first.Items) != 2 || first.NextCursor == "" {
t.Fatalf("first=%#v err=%v", first, err)
}
if err := store.SaveTask(ctx, root, snapshot("d", "session", 1, now.Add(4*time.Minute))); err != nil {
t.Fatal(err)
}
_ = catalog.Flush(flushCtx)
stale, err := catalog.ListPage(ctx, PageRequest{ProjectKeys: []string{project.Key}, Cursor: first.NextCursor})
if err != nil || !stale.StaleCursor || len(stale.Items) != 0 {
t.Fatalf("stale=%#v err=%v", stale, err)
}
}
+136
View File
@@ -0,0 +1,136 @@
package taskcatalog
import (
"context"
"path/filepath"
"sync"
"reasonix/internal/taskmonitor"
)
type sharedManager struct {
once sync.Once
mu sync.RWMutex
catalog *Catalog
pending map[string]string
closing bool
}
var shared sharedManager
func ensureShared() {
shared.once.Do(func() {
go func() {
catalog, err := Open(context.Background(), "")
if err != nil {
return
}
shared.mu.Lock()
if shared.closing {
shared.mu.Unlock()
_ = catalog.Close(context.Background())
return
}
shared.catalog = catalog
pending := shared.pending
shared.pending = nil
shared.mu.Unlock()
for root, label := range pending {
_, _ = catalog.RegisterProject(context.Background(), root, label)
}
}()
})
}
func Shared() *Catalog {
ensureShared()
shared.mu.RLock()
defer shared.mu.RUnlock()
return shared.catalog
}
// ShutdownShared drains accepted notifications and cancels every shared task
// projection worker. It is only used during process shutdown; authoritative
// task snapshots and event logs have already committed before notifications.
func ShutdownShared(ctx context.Context) error {
shared.mu.Lock()
shared.closing = true
catalog := shared.catalog
shared.catalog = nil
shared.mu.Unlock()
if catalog == nil {
return nil
}
flushErr := catalog.Flush(ctx)
closeErr := catalog.Close(ctx)
if flushErr != nil {
return flushErr
}
return closeErr
}
func RegisterSharedProject(root, label string) string {
ensureShared()
key := ProjectKey(root)
shared.mu.Lock()
if shared.closing {
shared.mu.Unlock()
return key
}
if shared.catalog == nil {
if shared.pending == nil {
shared.pending = map[string]string{}
}
shared.pending[root] = label
shared.mu.Unlock()
return key
}
catalog := shared.catalog
shared.mu.Unlock()
_, _ = catalog.RegisterProject(context.Background(), root, label)
return key
}
type sharedSink struct{}
func (sharedSink) SnapshotChanged(projectRoot, taskID string) {
catalog := sharedCatalogForNotification(projectRoot)
if catalog != nil {
catalog.SnapshotChanged(projectRoot, taskID)
}
}
func (sharedSink) EventsChanged(projectRoot, taskID string) {
catalog := sharedCatalogForNotification(projectRoot)
if catalog != nil {
catalog.EventsChanged(projectRoot, taskID)
}
}
// sharedCatalogForNotification is deliberately SQLite-free. ProjectionSink is
// called after the authoritative task file lock is released, but task saves
// still must never wait for catalog I/O. A notification received while the
// catalog is opening is recovered by the pending project's initial reconcile.
func sharedCatalogForNotification(projectRoot string) *Catalog {
ensureShared()
shared.mu.Lock()
defer shared.mu.Unlock()
if shared.closing {
return nil
}
if shared.catalog == nil {
if shared.pending == nil {
shared.pending = map[string]string{}
}
shared.pending[projectRoot] = filepath.Base(projectRoot)
return nil
}
return shared.catalog
}
// ObservedStore remains an authoritative FileStore; only its post-commit sink
// is shared with the disposable catalog.
func ObservedStore() taskmonitor.WriteStore {
ensureShared()
return taskmonitor.NewObservedFileStore(filepath.Join(".reasonix", "tasks"), sharedSink{})
}
+75
View File
@@ -0,0 +1,75 @@
package taskmonitor
import (
"bufio"
"context"
"encoding/json"
"errors"
"io"
"os"
"path/filepath"
)
type EventTail struct {
Items []TaskEvent
NextOffset int64
Reset bool
}
// ReadEventTail reads only complete JSONL lines after a catalog byte
// checkpoint. It preserves FileStore's identifier and symlink defenses.
func (s *FileStore) ReadEventTail(ctx context.Context, projectDir, taskID string, offset int64) (EventTail, error) {
out := EventTail{Items: []TaskEvent{}, NextOffset: offset}
if err := ctx.Err(); err != nil {
return out, err
}
id, err := safeID(taskID)
if err != nil {
return out, err
}
root, err := s.taskRoot(projectDir)
if err != nil {
return out, err
}
path := filepath.Join(root, id, "events.jsonl")
if err := rejectSymlinkChain(root, path); err != nil {
return out, err
}
f, err := os.Open(path)
if os.IsNotExist(err) {
return out, nil
}
if err != nil {
return out, err
}
defer f.Close()
info, err := f.Stat()
if err != nil {
return out, err
}
if offset < 0 || offset > info.Size() {
offset, out.NextOffset, out.Reset = 0, 0, true
}
if _, err := f.Seek(offset, io.SeekStart); err != nil {
return out, err
}
reader := bufio.NewReader(f)
for {
line, readErr := reader.ReadBytes('\n')
if len(line) > 0 && line[len(line)-1] == '\n' {
out.NextOffset += int64(len(line))
line = line[:len(line)-1]
var event TaskEvent
if json.Unmarshal(line, &event) == nil {
out.Items = append(out.Items, event)
}
}
if errors.Is(readErr, io.EOF) {
break
}
if readErr != nil {
return out, readErr
}
}
return out, nil
}
+49
View File
@@ -0,0 +1,49 @@
package taskmonitor
import (
"context"
"encoding/json"
"os"
"path/filepath"
"testing"
"time"
)
func TestReadEventTailKeepsIncompleteLineForRetry(t *testing.T) {
t.Parallel()
root := t.TempDir()
store := NewFileStore(filepath.Join(".reasonix", "tasks"))
now := time.Now()
first := TaskEvent{Timestamp: now, EventType: "state_change", TaskID: "task", State: TaskStateRunning}
if err := store.AppendAuditEvent(context.Background(), root, first); err != nil {
t.Fatal(err)
}
tail, err := store.ReadEventTail(context.Background(), root, "task", 0)
if err != nil || len(tail.Items) != 1 {
t.Fatalf("first tail=%#v err=%v", tail, err)
}
checkpoint := tail.NextOffset
second := TaskEvent{Sequence: 2, Timestamp: now.Add(time.Second), EventType: "state_change", TaskID: "task", State: TaskStateSucceeded}
line, _ := json.Marshal(second)
path := filepath.Join(root, ".reasonix", "tasks", "task", "events.jsonl")
f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o600)
if err != nil {
t.Fatal(err)
}
_, _ = f.Write(line)
_ = f.Close()
tail, err = store.ReadEventTail(context.Background(), root, "task", checkpoint)
if err != nil || len(tail.Items) != 0 || tail.NextOffset != checkpoint {
t.Fatalf("incomplete tail=%#v err=%v", tail, err)
}
f, err = os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o600)
if err != nil {
t.Fatal(err)
}
_, _ = f.Write([]byte{'\n'})
_ = f.Close()
tail, err = store.ReadEventTail(context.Background(), root, "task", checkpoint)
if err != nil || len(tail.Items) != 1 || tail.Items[0].Sequence != 2 {
t.Fatalf("completed tail=%#v err=%v", tail, err)
}
}
+19
View File
@@ -21,6 +21,7 @@ import (
// is added in TM-04.
type FileStore struct {
baseDir string // projectDir → task data root (e.g. ".reasonix/tasks")
sink ProjectionSink
}
// NewFileStore returns a FileStore rooted at baseDir. baseDir is typically
@@ -29,6 +30,10 @@ func NewFileStore(baseDir string) *FileStore {
return &FileStore{baseDir: baseDir}
}
func NewObservedFileStore(baseDir string, sink ProjectionSink) *FileStore {
return &FileStore{baseDir: baseDir, sink: sink}
}
// safeID validates a user-supplied identifier for use as a filesystem path
// component. It rejects empty strings, ".", "..", and values containing a
// path separator. Used for both taskID and idempotency keys.
@@ -332,6 +337,12 @@ func (s *FileStore) readEvents(taskDir string) ([]TaskEvent, error) {
// SaveTask implements WriteStore. It atomically writes the snapshot,
// failing if a concurrent write has changed the version.
func (s *FileStore) SaveTask(ctx context.Context, projectDir string, snap TaskSnapshot) (retErr error) {
committed := false
defer func() {
if committed && s.sink != nil {
s.sink.SnapshotChanged(projectDir, snap.TaskID)
}
}()
if err := ctx.Err(); err != nil {
return err
}
@@ -413,6 +424,7 @@ func (s *FileStore) SaveTask(ctx context.Context, projectDir string, snap TaskSn
return fmt.Errorf("save task: %w", err)
}
_ = os.Chmod(target, 0o600)
committed = true
return nil
}
@@ -420,6 +432,12 @@ func (s *FileStore) SaveTask(ctx context.Context, projectDir string, snap TaskSn
// AppendAuditEvent implements WriteStore. It atomically assigns the next
// monotonic sequence number and appends the event to the JSONL file.
func (s *FileStore) AppendAuditEvent(ctx context.Context, projectDir string, ev TaskEvent) (retErr error) {
committed := false
defer func() {
if committed && s.sink != nil {
s.sink.EventsChanged(projectDir, ev.TaskID)
}
}()
if err := ctx.Err(); err != nil {
return err
}
@@ -506,6 +524,7 @@ func (s *FileStore) AppendAuditEvent(ctx context.Context, projectDir string, ev
if _, err := f.WriteString(string(data) + "\n"); err != nil {
return err
}
committed = true
return nil
}
+4
View File
@@ -82,6 +82,10 @@ func reconcileRuntime(snap *TaskSnapshot, now time.Time) {
}
}
// ReconcileRuntime applies the read-time lease view without mutating the
// authoritative snapshot file.
func (ts *TaskSnapshot) ReconcileRuntime(now time.Time) { reconcileRuntime(ts, now) }
// ValidTaskStates is the set of well-known states.
var ValidTaskStates = map[TaskState]bool{
TaskStateQueued: true,
@@ -0,0 +1,56 @@
package taskmonitor
import (
"context"
"path/filepath"
"sync"
"testing"
"time"
)
type reentrantProjectionSink struct {
store *FileStore
mu sync.Mutex
ran bool
done chan error
}
func (s *reentrantProjectionSink) SnapshotChanged(root, taskID string) {
s.mu.Lock()
if s.ran {
s.mu.Unlock()
return
}
s.ran = true
s.mu.Unlock()
task, err := s.store.GetTask(context.Background(), root, taskID)
if err == nil && task != nil {
task.Version++
task.UpdatedAt = task.UpdatedAt.Add(time.Second)
err = s.store.SaveTask(context.Background(), root, *task)
}
s.done <- err
}
func (*reentrantProjectionSink) EventsChanged(string, string) {}
func TestProjectionSinkRunsAfterTaskLockRelease(t *testing.T) {
t.Parallel()
root := t.TempDir()
sink := &reentrantProjectionSink{done: make(chan error, 1)}
sink.store = NewObservedFileStore(filepath.Join(".reasonix", "tasks"), sink)
now := time.Now()
err := sink.store.SaveTask(context.Background(), root, TaskSnapshot{SchemaVersion: 1, TaskID: "task", State: TaskStateQueued,
Version: 1, CreatedAt: now, UpdatedAt: now})
if err != nil {
t.Fatal(err)
}
select {
case err := <-sink.done:
if err != nil {
t.Fatal(err)
}
case <-time.After(2 * time.Second):
t.Fatal("projection sink was invoked while the task lock was still held")
}
}
+7
View File
@@ -18,6 +18,13 @@ type Store interface {
ListEvents(ctx context.Context, projectDir string, taskID string, afterSequence int) ([]TaskEvent, error)
}
// ProjectionSink receives post-commit hints. Implementations must enqueue and
// return immediately; FileStore remains the only authority for task control.
type ProjectionSink interface {
SnapshotChanged(projectRoot, taskID string)
EventsChanged(projectRoot, taskID string)
}
// IdempotencyRecord captures the binding between an idempotency key and the
// operation it was used for.
type IdempotencyRecord struct {