feat(subconscious): stabilize heartbeat + subconscious loop (#392) (#437)

* feat(subconscious): stabilize heartbeat + subconscious loop (#392)

- Enable heartbeat by default (enabled=true, inference_enabled=true, 5min interval)
- Seed system tasks on engine init, not first tick
- SQLite-backed task/log/escalation persistence
- Overlap guard with generation counter — stale ticks are cancelled
- Single log entry per task per tick, updated in place (in_progress → act/noop/escalate/failed/cancelled)
- Rate-limit retry (429 only) for agentic-v1 cloud model calls
- Approval gate: unsolicited write actions on read-only tasks require user approval
- Analysis-only mode for agentic-v1 on read-only escalations
- Non-blocking status RPC — reads from DB, never blocks on engine mutex
- Frontend: system vs user task distinction, toggle switches, expandable activity log
- Frontend: 3s auto-poll on Subconscious tab, skill-related escalation navigation
- Consecutive failure counter in status (resets on success)
- last_tick_at only advances on successful evaluation
- Missing LLM evaluation fallback — unevaluated tasks default to noop
- Docs: subconscious.md architecture guide, memory-sync-functions.md reference

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* style: fix Prettier formatting for subconscious frontend files

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* ci: retrigger checks

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(heartbeat): use disabled config in run_returns_immediately_when_disabled test

HeartbeatConfig::default() has enabled: true, so run() entered the infinite
loop and never returned — hanging the test (and CI) indefinitely.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(subconscious): remove HEARTBEAT.md task import, use SQLite as sole task source

Tasks are now managed exclusively in SQLite via the Subconscious UI.
HEARTBEAT.md is retained for instructions/context only, not as a task list.
Situation report now reads pending tasks from SQLite instead of HEARTBEAT.md.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* style: cargo fmt on subconscious engine

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
sanil-23
2026-04-08 15:03:05 -07:00
committed by GitHub
co-authored by Claude Opus 4.6
parent 1ca4ea044a
commit fa5f822f95
19 changed files with 4056 additions and 995 deletions
+205
View File
@@ -0,0 +1,205 @@
/**
* useSubconscious — hook for the subconscious engine UI.
*
* Provides tasks, escalations, execution log, and actions for the
* subconscious tab on the Intelligence page.
*/
import { useCallback, useEffect, useRef, useState } from 'react';
import {
isTauri,
subconsciousEscalationsApprove,
subconsciousEscalationsDismiss,
subconsciousEscalationsList,
subconsciousLogList,
subconsciousStatus,
subconsciousTasksAdd,
subconsciousTasksList,
subconsciousTasksRemove,
subconsciousTasksUpdate,
subconsciousTrigger,
} from '../utils/tauriCommands';
import type {
SubconsciousEscalation,
SubconsciousLogEntry,
SubconsciousStatus,
SubconsciousTask,
} from '../utils/tauriCommands/subconscious';
export interface UseSubconsciousResult {
// Data
tasks: SubconsciousTask[];
escalations: SubconsciousEscalation[];
logEntries: SubconsciousLogEntry[];
status: SubconsciousStatus | null;
// Loading states
loading: boolean;
triggering: boolean;
// Actions
refresh: () => Promise<void>;
triggerTick: () => Promise<void>;
addTask: (title: string) => Promise<void>;
removeTask: (taskId: string) => Promise<void>;
toggleTask: (taskId: string, enabled: boolean) => Promise<void>;
approveEscalation: (escalationId: string) => Promise<void>;
dismissEscalation: (escalationId: string) => Promise<void>;
// Error
error: string | null;
}
export function useSubconscious(): UseSubconsciousResult {
const [tasks, setTasks] = useState<SubconsciousTask[]>([]);
const [escalations, setEscalations] = useState<SubconsciousEscalation[]>([]);
const [logEntries, setLogEntries] = useState<SubconsciousLogEntry[]>([]);
const [status, setStatus] = useState<SubconsciousStatus | null>(null);
const [loading, setLoading] = useState(false);
const [triggering, setTriggering] = useState(false);
const [error, setError] = useState<string | null>(null);
const fetchingRef = useRef(false);
const refresh = useCallback(async () => {
if (!isTauri() || fetchingRef.current) return;
fetchingRef.current = true;
setLoading(true);
setError(null);
try {
const [tasksRes, escalationsRes, logRes, statusRes] = await Promise.all([
subconsciousTasksList().catch(() => null),
subconsciousEscalationsList('pending').catch(() => null),
subconsciousLogList(undefined, 30).catch(() => null),
subconsciousStatus().catch(() => null),
]);
if (tasksRes) setTasks(unwrap(tasksRes) ?? []);
if (escalationsRes) setEscalations(unwrap(escalationsRes) ?? []);
if (logRes) setLogEntries(unwrap(logRes) ?? []);
if (statusRes) setStatus(unwrap(statusRes) ?? null);
} catch (err) {
setError(err instanceof Error ? err.message : 'Failed to load subconscious data');
} finally {
setLoading(false);
fetchingRef.current = false;
}
}, []);
const triggerTick = useCallback(async () => {
if (!isTauri() || triggering) return;
setTriggering(true);
try {
await subconsciousTrigger();
} catch (err) {
console.warn('[subconscious] trigger failed:', err);
} finally {
setTriggering(false);
}
}, [triggering]);
const addTask = useCallback(
async (title: string) => {
if (!isTauri()) return;
try {
await subconsciousTasksAdd(title);
await refresh();
} catch (err) {
console.warn('[subconscious] add task failed:', err);
throw err;
}
},
[refresh]
);
const removeTask = useCallback(
async (taskId: string) => {
if (!isTauri()) return;
try {
await subconsciousTasksRemove(taskId);
await refresh();
} catch (err) {
console.warn('[subconscious] remove task failed:', err);
}
},
[refresh]
);
const toggleTask = useCallback(
async (taskId: string, enabled: boolean) => {
if (!isTauri()) return;
try {
await subconsciousTasksUpdate(taskId, { enabled });
await refresh();
} catch (err) {
console.warn('[subconscious] toggle task failed:', err);
}
},
[refresh]
);
const approveEscalation = useCallback(
async (escalationId: string) => {
if (!isTauri()) return;
try {
await subconsciousEscalationsApprove(escalationId);
await refresh();
} catch (err) {
console.warn('[subconscious] approve failed:', err);
throw err;
}
},
[refresh]
);
const dismissEscalation = useCallback(
async (escalationId: string) => {
if (!isTauri()) return;
try {
await subconsciousEscalationsDismiss(escalationId);
await refresh();
} catch (err) {
console.warn('[subconscious] dismiss failed:', err);
}
},
[refresh]
);
// Poll every 3s while the hook is mounted (user is on Subconscious tab).
// Picks up all state changes: in_progress → act/noop/escalate/failed,
// new escalations, background tick completions, etc.
useEffect(() => {
refresh();
const interval = setInterval(refresh, 3000);
return () => clearInterval(interval);
}, [refresh]);
return {
tasks,
escalations,
logEntries,
status,
loading,
triggering,
refresh,
triggerTick,
addTask,
removeTask,
toggleTask,
approveEscalation,
dismissEscalation,
error,
};
}
/**
* Unwrap a CommandResponse — callCoreRpc returns `{ result: T, logs: [...] }`.
*/
function unwrap<T>(response: unknown): T | null {
if (!response || typeof response !== 'object') return null;
const r = response as Record<string, unknown>;
// CommandResponse shape: { result: T, logs: string[] }
if ('result' in r) {
return r.result as T;
}
return null;
}
+375 -30
View File
@@ -1,4 +1,5 @@
import { useCallback, useEffect, useMemo, useState } from 'react';
import { useNavigate } from 'react-router-dom';
import { ActionableCard } from '../components/intelligence/ActionableCard';
import { ConfirmationModal } from '../components/intelligence/ConfirmationModal';
@@ -15,6 +16,7 @@ import {
} from '../hooks/useIntelligenceSocket';
import { useIntelligenceStats } from '../hooks/useIntelligenceStats';
import { useScreenIntelligenceItems } from '../hooks/useScreenIntelligenceItems';
import { useSubconscious } from '../hooks/useSubconscious';
import type {
ActionableItem,
ActionableItemSource,
@@ -25,7 +27,15 @@ import type {
type IntelligenceTab = 'memory' | 'subconscious' | 'dreams';
const SKILL_KEYWORDS =
/\bskill\b|\boauth\b|\bnotion\b|\bgmail\b|\bintegration\b|\bdisconnect|\breconnect|\bre-?auth/i;
function isSkillRelated(title: string, description: string): boolean {
return SKILL_KEYWORDS.test(title) || SKILL_KEYWORDS.test(description);
}
export default function Intelligence() {
const navigate = useNavigate();
const { aiStatus } = useIntelligenceStats();
const [activeTab, setActiveTab] = useState<IntelligenceTab>('memory');
@@ -45,6 +55,24 @@ export default function Intelligence() {
const { mutateAsync: updateItemStatus } = useUpdateActionableItem();
const { mutateAsync: snoozeItem } = useSnoozeActionableItem();
// Subconscious engine data
const {
tasks: subconsciousTasks,
escalations,
logEntries,
status: subconsciousEngineStatus,
loading: subconsciousLoading,
triggering: subconsciousTriggering,
triggerTick,
addTask: addSubconsciousTask,
removeTask: removeSubconsciousTask,
toggleTask: toggleSubconsciousTask,
approveEscalation,
dismissEscalation,
} = useSubconscious();
const [newTaskTitle, setNewTaskTitle] = useState('');
const [expandedLogIds, setExpandedLogIds] = useState<Set<string>>(new Set());
// Socket integration
const socketManager = useIntelligenceSocketManager();
const { isConnected: socketConnected } = useIntelligenceSocket();
@@ -239,7 +267,7 @@ export default function Intelligence() {
const tabs: { id: IntelligenceTab; label: string; comingSoon?: boolean }[] = [
{ id: 'memory', label: 'Memory' },
{ id: 'subconscious', label: 'Subconscious', comingSoon: true },
{ id: 'subconscious', label: 'Subconscious' },
{ id: 'dreams', label: 'Dreams', comingSoon: true },
];
@@ -259,10 +287,12 @@ export default function Intelligence() {
)}
</div>
<div className="flex items-center gap-3">
<div className="flex items-center gap-2">
<div className={`w-2 h-2 rounded-full ${systemStatusDot}`} />
<span className="text-xs text-stone-400">{systemStatusLabel}</span>
</div>
{activeTab === 'memory' && (
<div className="flex items-center gap-2">
<div className={`w-2 h-2 rounded-full ${systemStatusDot}`} />
<span className="text-xs text-stone-400">{systemStatusLabel}</span>
</div>
)}
{activeTab === 'memory' && (
<button
onClick={usingMemoryData ? refreshConscious : handleAnalyzeNow}
@@ -454,33 +484,348 @@ export default function Intelligence() {
)}
{activeTab === 'subconscious' && (
<div className="glass rounded-2xl p-8 text-center animate-fade-up">
<div className="w-16 h-16 mx-auto mb-4 flex items-center justify-center rounded-full bg-lavender-500/10">
<svg
className="w-8 h-8 text-lavender-400"
fill="none"
stroke="currentColor"
viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={1.5}
d="M21 12a9 9 0 11-18 0 9 9 0 0118 0z"
<div className="space-y-6 animate-fade-up">
{/* Status bar */}
<div className="flex items-center justify-between">
<div className="flex items-center gap-2 text-xs text-stone-400">
{subconsciousEngineStatus && (
<>
<span>{subconsciousEngineStatus.task_count} tasks</span>
<span className="text-stone-300">|</span>
<span>{subconsciousEngineStatus.total_ticks} ticks</span>
{subconsciousEngineStatus.last_tick_at && (
<>
<span className="text-stone-300">|</span>
<span>
Last:{' '}
{new Date(
subconsciousEngineStatus.last_tick_at * 1000
).toLocaleTimeString()}
</span>
</>
)}
{subconsciousEngineStatus.consecutive_failures > 0 && (
<>
<span className="text-stone-300">|</span>
<span className="text-coral-500">
{subconsciousEngineStatus.consecutive_failures} failed
</span>
</>
)}
</>
)}
</div>
<div className="flex items-center gap-2">
{/* Interval selector */}
<div className="flex items-center gap-1.5">
<svg
className="w-3 h-3 text-stone-400"
fill="none"
stroke="currentColor"
viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={2}
d="M12 8v4l3 3m6-3a9 9 0 11-18 0 9 9 0 0118 0z"
/>
</svg>
<select
value={subconsciousEngineStatus?.interval_minutes ?? 5}
onChange={() => {
// Config update would require restart — show as read-only for now
}}
disabled
title="Tick interval (change in Settings > Advanced)"
className="text-xs bg-stone-50 border border-stone-200 rounded px-1.5 py-0.5 text-stone-500 cursor-not-allowed">
<option value={5}>5 min</option>
<option value={10}>10 min</option>
<option value={15}>15 min</option>
<option value={30}>30 min</option>
<option value={60}>1 hour</option>
<option value={360}>6 hours</option>
<option value={720}>12 hours</option>
<option value={1440}>1 day</option>
</select>
</div>
<button
onClick={triggerTick}
disabled={subconsciousTriggering}
className="flex items-center gap-1.5 px-3 py-1.5 text-xs bg-stone-50 hover:bg-stone-100 disabled:opacity-40 border border-stone-200 rounded-lg text-stone-600 transition-colors">
{subconsciousTriggering ? (
<div className="w-3 h-3 border border-stone-400 border-t-transparent rounded-full animate-spin" />
) : (
<svg
className="w-3 h-3"
fill="none"
stroke="currentColor"
viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={2}
d="M13 10V3L4 14h7v7l9-11h-7z"
/>
</svg>
)}
Run Now
</button>
</div>
</div>
{/* Escalations — needs user input */}
{escalations.length > 0 && (
<div>
<h3 className="text-sm font-semibold text-stone-900 mb-3 flex items-center gap-2">
<span className="w-2 h-2 rounded-full bg-amber-400 animate-pulse" />
Approval Needed
<span className="text-[10px] px-1.5 py-0.5 rounded-full bg-amber-100 text-amber-700">
{escalations.length}
</span>
</h3>
<div className="space-y-2">
{escalations.map(esc => (
<div
key={esc.id}
className="bg-amber-50 border border-amber-200 rounded-xl p-4">
<div className="flex items-start justify-between">
<div className="flex-1">
<p className="text-sm font-medium text-stone-900">{esc.title}</p>
<p className="text-xs text-stone-500 mt-1">{esc.description}</p>
<div className="flex items-center gap-2 mt-2">
<span
className={`text-[10px] px-2 py-0.5 rounded-full ${
esc.priority === 'critical'
? 'bg-coral-100 text-coral-700'
: esc.priority === 'important'
? 'bg-amber-100 text-amber-700'
: 'bg-stone-100 text-stone-600'
}`}>
{esc.priority}
</span>
<span className="text-[10px] text-stone-400">
Requires your approval to proceed
</span>
</div>
</div>
<div className="flex gap-2 ml-3 flex-shrink-0">
{isSkillRelated(esc.title, esc.description) ? (
<button
onClick={() => {
dismissEscalation(esc.id);
navigate('/skills');
}}
className="px-3 py-1.5 text-xs bg-primary-500 hover:bg-primary-600 text-white rounded-lg transition-colors">
Fix in Skills
</button>
) : (
<button
onClick={() => approveEscalation(esc.id)}
className="px-3 py-1.5 text-xs bg-sage-500 hover:bg-sage-600 text-white rounded-lg transition-colors">
Go ahead
</button>
)}
<button
onClick={() => dismissEscalation(esc.id)}
className="px-3 py-1.5 text-xs bg-stone-100 hover:bg-stone-200 text-stone-600 rounded-lg transition-colors">
Skip
</button>
</div>
</div>
</div>
))}
</div>
</div>
)}
{/* Active tasks */}
<div>
<h3 className="text-sm font-semibold text-stone-900 mb-3">Active Tasks</h3>
{subconsciousLoading && subconsciousTasks.length === 0 ? (
<div className="text-center py-4">
<div className="w-6 h-6 mx-auto border-2 border-stone-300 border-t-transparent rounded-full animate-spin" />
</div>
) : subconsciousTasks.filter(t => !t.completed).length === 0 ? (
<p className="text-xs text-stone-400 py-3">No active tasks. Add one below.</p>
) : (
<div className="space-y-1.5">
{/* System tasks — always-on, no controls */}
{subconsciousTasks
.filter(t => !t.completed && t.source === 'system')
.map(task => (
<div
key={task.id}
className="flex items-center py-2 px-3 bg-stone-50 rounded-lg">
<div className="w-1.5 h-1.5 rounded-full bg-sage-400 flex-shrink-0 mr-2.5" />
<span className="text-sm text-stone-900 truncate flex-1">
{task.title}
</span>
<span className="text-[10px] text-stone-400 flex-shrink-0 px-1.5 py-0.5 rounded bg-stone-100">
default
</span>
</div>
))}
{/* User tasks — toggle switch + delete */}
{subconsciousTasks
.filter(t => !t.completed && t.source !== 'system')
.map(task => (
<div
key={task.id}
className="flex items-center justify-between py-2 px-3 bg-stone-50 rounded-lg group">
<div className="flex items-center gap-2.5 flex-1 min-w-0">
<button
onClick={() => toggleSubconsciousTask(task.id, !task.enabled)}
className={`relative w-7 h-4 rounded-full flex-shrink-0 transition-colors ${
task.enabled ? 'bg-sage-500' : 'bg-stone-300'
}`}>
<span
className={`absolute top-0.5 left-0.5 w-3 h-3 rounded-full bg-white shadow transition-transform ${
task.enabled ? 'translate-x-3' : 'translate-x-0'
}`}
/>
</button>
<span
className={`text-sm truncate ${task.enabled ? 'text-stone-900' : 'text-stone-400'}`}>
{task.title}
</span>
</div>
<button
onClick={() => removeSubconsciousTask(task.id)}
className="opacity-0 group-hover:opacity-100 p-1 text-stone-400 hover:text-coral-500 transition-all">
<svg
className="w-3.5 h-3.5"
fill="none"
stroke="currentColor"
viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={2}
d="M6 18L18 6M6 6l12 12"
/>
</svg>
</button>
</div>
))}
</div>
)}
{/* Add task */}
<form
onSubmit={async e => {
e.preventDefault();
const title = newTaskTitle.trim();
if (!title) return;
try {
await addSubconsciousTask(title);
setNewTaskTitle('');
} catch {
// handled by hook
}
}}
className="flex gap-2 mt-3">
<input
type="text"
placeholder="Add a task... (e.g. 'Check urgent emails')"
value={newTaskTitle}
onChange={e => setNewTaskTitle(e.target.value)}
className="flex-1 px-3 py-2 text-sm bg-white border border-stone-200 rounded-lg text-stone-900 placeholder-stone-400 focus:outline-none focus:border-primary-500/50 transition-colors"
/>
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={1.5}
d="M9 9.563C9 9.252 9.252 9 9.563 9h4.874c.311 0 .563.252.563.563v4.874c0 .311-.252.563-.563.563H9.564A.562.562 0 019 14.437V9.564z"
/>
</svg>
<button
type="submit"
disabled={!newTaskTitle.trim()}
className="px-3 py-2 text-sm bg-primary-500 hover:bg-primary-600 disabled:opacity-40 text-white rounded-lg transition-colors">
Add
</button>
</form>
</div>
{/* Execution log */}
<div>
<h3 className="text-sm font-semibold text-stone-900 mb-3">Activity Log</h3>
{logEntries.length === 0 ? (
<p className="text-xs text-stone-400 py-3">
No activity yet. Run a tick to see results.
</p>
) : (
<div className="space-y-1 max-h-64 overflow-y-auto">
{logEntries.map(entry => (
<div key={entry.id} className="flex items-start gap-2 py-1.5 px-2 text-xs">
<span className="text-stone-400 flex-shrink-0 w-14">
{new Date(entry.tick_at * 1000).toLocaleTimeString([], {
hour: '2-digit',
minute: '2-digit',
})}
</span>
<span
className={`flex-shrink-0 w-1.5 h-1.5 rounded-full mt-1.5 ${
entry.decision === 'act'
? 'bg-sage-400'
: entry.decision === 'in_progress'
? 'bg-primary-400 animate-pulse'
: entry.decision === 'escalate'
? 'bg-amber-400'
: entry.decision === 'failed'
? 'bg-coral-400'
: entry.decision === 'cancelled'
? 'bg-stone-300'
: entry.decision === 'dismissed'
? 'bg-stone-300'
: 'bg-stone-200'
}`}
/>
<span
className={`break-words min-w-0 ${
entry.decision === 'in_progress'
? 'text-stone-400'
: entry.decision === 'failed'
? 'text-coral-500'
: 'text-stone-600'
} ${entry.result && entry.result.length > 120 ? 'cursor-pointer hover:text-stone-900' : ''}`}
onClick={() => {
if (entry.result && entry.result.length > 120) {
setExpandedLogIds(prev => {
const next = new Set(prev);
if (next.has(entry.id)) next.delete(entry.id);
else next.add(entry.id);
return next;
});
}
}}>
{entry.result
? expandedLogIds.has(entry.id)
? entry.result
: entry.result.length > 120
? `${entry.result.substring(0, 120)}...`
: entry.result
: entry.decision === 'noop'
? 'Nothing new'
: entry.decision === 'act'
? 'Completed'
: entry.decision === 'in_progress'
? 'Evaluating...'
: entry.decision === 'escalate'
? 'Waiting for approval'
: entry.decision === 'failed'
? 'Failed'
: entry.decision === 'cancelled'
? 'Cancelled'
: entry.decision === 'dismissed'
? 'Skipped'
: entry.decision}
</span>
{entry.duration_ms != null && (
<span className="text-stone-300 flex-shrink-0 ml-auto">
{entry.duration_ms > 1000
? `${(entry.duration_ms / 1000).toFixed(1)}s`
: `${entry.duration_ms}ms`}
</span>
)}
</div>
))}
</div>
)}
</div>
<h2 className="text-lg font-semibold text-stone-900 mb-2">Subconscious</h2>
<p className="text-stone-400 text-sm mb-1">
OpenHuman will constantly have subconscious thoughts based on all the information
it has access to and the activity you have engaged with it in.
</p>
<p className="text-xs text-stone-500">Coming soon</p>
</div>
)}
+1
View File
@@ -8,6 +8,7 @@ export * from './core';
export * from './memory';
export * from './webhooks';
export * from './conscious';
export * from './subconscious';
export * from './localAi';
export * from './config';
export * from './cron';
+164
View File
@@ -0,0 +1,164 @@
/**
* Subconscious engine commands — task management, escalations, execution log.
*/
import { callCoreRpc } from '../../services/coreRpcClient';
import { type CommandResponse, isTauri } from './common';
// ── Types ────────────────────────────────────────────────────────────────────
export interface SubconsciousTask {
id: string;
title: string;
source: 'system' | 'user';
recurrence: string;
enabled: boolean;
last_run_at: number | null;
next_run_at: number | null;
completed: boolean;
created_at: number;
}
export interface SubconsciousLogEntry {
id: string;
task_id: string;
tick_at: number;
decision: 'noop' | 'act' | 'escalate' | 'dismissed' | string;
result: string | null;
duration_ms: number | null;
created_at: number;
}
export interface SubconsciousEscalation {
id: string;
task_id: string;
log_id: string | null;
title: string;
description: string;
priority: 'critical' | 'important' | 'normal';
status: 'pending' | 'approved' | 'dismissed';
created_at: number;
resolved_at: number | null;
}
export interface SubconsciousStatus {
enabled: boolean;
interval_minutes: number;
last_tick_at: number | null;
total_ticks: number;
task_count: number;
pending_escalations: number;
consecutive_failures: number;
}
export interface TickResult {
tick_at: number;
evaluations: Array<{ task_id: string; decision: string; reason: string }>;
executed: number;
escalated: number;
duration_ms: number;
}
// ── Status & Trigger ─────────────────────────────────────────────────────────
export async function subconsciousStatus(): Promise<CommandResponse<SubconsciousStatus>> {
if (!isTauri()) throw new Error('Not running in Tauri');
return await callCoreRpc<CommandResponse<SubconsciousStatus>>({
method: 'openhuman.subconscious_status',
});
}
export async function subconsciousTrigger(): Promise<CommandResponse<TickResult>> {
if (!isTauri()) throw new Error('Not running in Tauri');
return await callCoreRpc<CommandResponse<TickResult>>({
method: 'openhuman.subconscious_trigger',
});
}
// ── Tasks CRUD ───────────────────────────────────────────────────────────────
export async function subconsciousTasksList(
enabledOnly = false
): Promise<CommandResponse<SubconsciousTask[]>> {
if (!isTauri()) throw new Error('Not running in Tauri');
return await callCoreRpc<CommandResponse<SubconsciousTask[]>>({
method: 'openhuman.subconscious_tasks_list',
params: { enabled_only: enabledOnly },
});
}
export async function subconsciousTasksAdd(
title: string,
source: 'user' | 'system' = 'user'
): Promise<CommandResponse<SubconsciousTask>> {
if (!isTauri()) throw new Error('Not running in Tauri');
return await callCoreRpc<CommandResponse<SubconsciousTask>>({
method: 'openhuman.subconscious_tasks_add',
params: { title, source },
});
}
export async function subconsciousTasksUpdate(
taskId: string,
patch: { title?: string; enabled?: boolean }
): Promise<CommandResponse<{ updated: string }>> {
if (!isTauri()) throw new Error('Not running in Tauri');
return await callCoreRpc<CommandResponse<{ updated: string }>>({
method: 'openhuman.subconscious_tasks_update',
params: { task_id: taskId, ...patch },
});
}
export async function subconsciousTasksRemove(
taskId: string
): Promise<CommandResponse<{ removed: string }>> {
if (!isTauri()) throw new Error('Not running in Tauri');
return await callCoreRpc<CommandResponse<{ removed: string }>>({
method: 'openhuman.subconscious_tasks_remove',
params: { task_id: taskId },
});
}
// ── Log ──────────────────────────────────────────────────────────────────────
export async function subconsciousLogList(
taskId?: string,
limit = 50
): Promise<CommandResponse<SubconsciousLogEntry[]>> {
if (!isTauri()) throw new Error('Not running in Tauri');
return await callCoreRpc<CommandResponse<SubconsciousLogEntry[]>>({
method: 'openhuman.subconscious_log_list',
params: { task_id: taskId, limit },
});
}
// ── Escalations ──────────────────────────────────────────────────────────────
export async function subconsciousEscalationsList(
status?: 'pending' | 'approved' | 'dismissed'
): Promise<CommandResponse<SubconsciousEscalation[]>> {
if (!isTauri()) throw new Error('Not running in Tauri');
return await callCoreRpc<CommandResponse<SubconsciousEscalation[]>>({
method: 'openhuman.subconscious_escalations_list',
params: status ? { status } : {},
});
}
export async function subconsciousEscalationsApprove(
escalationId: string
): Promise<CommandResponse<{ approved: string }>> {
if (!isTauri()) throw new Error('Not running in Tauri');
return await callCoreRpc<CommandResponse<{ approved: string }>>({
method: 'openhuman.subconscious_escalations_approve',
params: { escalation_id: escalationId },
});
}
export async function subconsciousEscalationsDismiss(
escalationId: string
): Promise<CommandResponse<{ dismissed: string }>> {
if (!isTauri()) throw new Error('Not running in Tauri');
return await callCoreRpc<CommandResponse<{ dismissed: string }>>({
method: 'openhuman.subconscious_escalations_dismiss',
params: { escalation_id: escalationId },
});
}
+4
View File
@@ -198,7 +198,11 @@ let relations = client.graph_query(None, None, None).await?;
|----------|-------------|
| `delete_document(namespace, id)` | Remove a specific document |
| `kv_delete(namespace, key)` | Remove a specific KV entry |
<<<<<<< HEAD
| `clear_skill_memory(skill_id, integration_id)` | Disconnect / revoke: clears skill-scoped memory in the shared `skill-{skill_id}` namespace. Storage is not isolated per integration—multiple integrations share that namespace; `integration_id` identifies the integration in the API contract (see implementation in `MemoryClient::clear_skill_memory`) |
=======
| `clear_skill_memory(skill_id, integration_id)` | Skill OAuth/auth revoked — wipes `skill-{skill_id}` namespace |
>>>>>>> 5b6b1e2 (feat(subconscious): stabilize heartbeat + subconscious loop (#392))
| `clear_namespace(namespace)` | Wipe an arbitrary namespace |
---
+309
View File
@@ -0,0 +1,309 @@
# Subconscious Loop
Background task evaluation and execution system. Periodically checks user-defined and system tasks against the current workspace state, decides what to do, and either acts autonomously or escalates to the user.
---
## Architecture
```
┌─────────────────────────────────────────────────────────┐
│ Heartbeat Engine │
│ (sleeps N minutes between ticks) │
└──────────────────────┬──────────────────────────────────┘
┌─────────────────────────────────────────────────────────┐
│ Subconscious Engine │
│ │
│ 1. Load due tasks from SQLite │
│ 2. Insert in_progress log entries │
│ 3. Build situation report (memory + workspace state) │
│ 4. Evaluate tasks with local Ollama model │
│ 5. Execute decisions (act / noop / escalate) │
│ 6. Update log entries in place │
└─────────────────────────────────────────────────────────┘
┌───────────┼───────────┐
▼ ▼ ▼
noop act escalate
(skip) (execute) (agentic-v1)
```
### Key files
| File | Purpose |
|------|---------|
| `src/openhuman/heartbeat/engine.rs` | Periodic scheduler, delegates to subconscious engine |
| `src/openhuman/subconscious/engine.rs` | Core tick logic, state management, overlap guard |
| `src/openhuman/subconscious/executor.rs` | Task execution routing (local model vs agentic-v1) |
| `src/openhuman/subconscious/prompt.rs` | Prompt builders for evaluation and execution |
| `src/openhuman/subconscious/store.rs` | SQLite persistence (tasks, log, escalations) |
| `src/openhuman/subconscious/types.rs` | Data types and enums |
| `src/openhuman/subconscious/situation_report.rs` | Builds context from memory and workspace state |
| `src/openhuman/subconscious/global.rs` | Global singleton shared between heartbeat and RPC |
| `src/openhuman/subconscious/schemas.rs` | RPC endpoint handlers |
| `app/src/hooks/useSubconscious.ts` | Frontend hook for data fetching and actions |
| `app/src/pages/Intelligence.tsx` | UI rendering (Subconscious tab) |
| `app/src/utils/tauriCommands/subconscious.ts` | TypeScript RPC wrappers |
---
## Task Types
### System tasks
Seeded automatically on engine initialization. Cannot be deleted, only disabled.
Default system tasks:
- Check connected skills for errors or disconnections
- Review new memory updates for actionable items
- Monitor system health (Ollama, memory, connections)
Additional system tasks are imported from `HEARTBEAT.md` in the workspace directory (one task per `- ` line).
### User tasks
Added manually via the UI. Can be toggled on/off and deleted.
Examples:
- "Check urgent emails" (read-only)
- "Send daily summary to Slack" (write intent)
- "Summarize Notion updates" (read-only)
---
## Tick Lifecycle
### 1. Overlap guard
Each tick gets a monotonically increasing generation counter. If a new tick starts while the old one is still running (e.g., slow LLM call), the old tick's results are discarded and its `in_progress` log entries are marked as `cancelled`.
The heartbeat uses `tokio::time::sleep` (not `interval`) so ticks never stack up.
### 2. Load due tasks
Query: enabled, not completed, `next_run_at <= now` or never run.
### 3. Log as in_progress
Each due task gets a single log row inserted with `decision = "in_progress"`. This row is updated in place as the task progresses — no duplicate rows.
### 4. Evaluate with local model
The local Ollama model receives all due tasks + a situation report and returns a per-task decision:
| Decision | Meaning |
|----------|---------|
| `noop` | Nothing relevant right now |
| `act` | Something relevant found — execute the task |
| `escalate` | Needs deeper reasoning — hand off to agentic-v1 |
### 5. Execute
Routing depends on the decision and the task's intent:
```
Decision: noop
→ Update log to "noop", advance schedule
Decision: act
├─ Task has write intent (needs_tools = true)
│ → Execute with local model
└─ Task is read-only
→ Execute with local model
Decision: escalate
├─ Task has write intent (needs_tools = true)
│ → Run agentic-v1 with full permissions
│ → No approval needed (user explicitly asked for the write action)
└─ Task is read-only
→ Run agentic-v1 in analysis-only mode
→ If response contains "RECOMMENDED ACTION:" (unsolicited write)
│ → Create escalation for user approval
│ → On approval → run agentic-v1 with full permissions
└─ Otherwise → log result, done
```
### 6. Update log entry
The same row inserted in step 3 is updated to the final state:
| Decision | Dot color | Text |
|----------|-----------|------|
| `in_progress` | Blue (pulsing) | "Evaluating..." |
| `act` | Green | Result text |
| `noop` | Gray | "Nothing new" |
| `escalate` | Amber | "Waiting for approval" |
| `failed` | Coral | Error message |
| `cancelled` | Gray | "Cancelled" |
| `dismissed` | Gray | "Skipped" |
---
## Execution Models
### Local Ollama model
Used for:
- Task evaluation (all tasks, every tick)
- Text-only task execution (summarize, check, monitor, review)
No cost, no rate limits, runs on-device.
### agentic-v1 (cloud)
Used for:
- Tool-required task execution (send, post, delete, create)
- Analysis-only mode for read-only tasks escalated by the local model
Rate-limit retry: up to 3 attempts with exponential backoff (2s, 4s, 8s) on 429 errors.
---
## Approval Gate
Approval is only required when the AI wants to take a **write action that the user didn't explicitly request**.
| Task intent | AI wants to write | Approval needed? |
|-------------|-------------------|-----------------|
| "Send digest to Slack" (write) | Yes | No — user asked for it |
| "Check urgent emails" (read) | No | No — read-only result |
| "Check urgent emails" (read) | Yes (wants to forward them) | **Yes** — unsolicited write |
The approval flow:
1. agentic-v1 runs in analysis-only mode
2. Response contains `RECOMMENDED ACTION: Forward 3 urgent emails to #team-alerts`
3. Escalation card appears in UI under "Approval Needed"
4. User clicks "Go ahead" → agentic-v1 runs again with full permissions
5. Or user clicks "Skip" → nothing happens
### Skill-related escalations
Escalations related to skills (detected by keywords: skill, oauth, notion, gmail, integration, disconnect, re-auth) show a "Fix in Skills" button that navigates to the Skills page instead of "Go ahead".
---
## Failure Handling
### Consecutive failure counter
Tracked in `EngineState.consecutive_failures`. Increments when the entire LLM evaluation fails (Ollama down, network error). Resets to 0 on any successful tick. Surfaced in the UI status bar as "N failed" in coral.
Individual task execution failures do NOT increment this counter — they are logged per-task but the tick itself is considered successful.
### last_tick_at advancement
`last_tick_at` only advances on successful ticks. If the LLM evaluation fails or the tick is cancelled, `last_tick_at` stays unchanged so the next tick's situation report covers the same time range — nothing is missed.
---
## Configuration
In `config.toml` under `[heartbeat]`:
```toml
[heartbeat]
enabled = true # Enable the heartbeat loop
interval_minutes = 5 # Tick interval (minimum 5)
inference_enabled = true # Enable local model evaluation
context_budget_tokens = 40000 # Max tokens for situation report
```
Defaults: `enabled = true`, `interval_minutes = 5`, `inference_enabled = true`.
---
## SQLite Schema
Database: `{workspace_dir}/subconscious/subconscious.db`
### subconscious_tasks
| Column | Type | Description |
|--------|------|-------------|
| id | TEXT PK | UUID |
| title | TEXT | Task description |
| source | TEXT | `"system"` or `"user"` |
| recurrence | TEXT | `"pending"`, `"once"`, or `"cron:expr"` |
| enabled | INTEGER | 1 = active, 0 = paused |
| last_run_at | REAL | Unix timestamp of last evaluation |
| next_run_at | REAL | Unix timestamp of next scheduled run |
| completed | INTEGER | 1 = done (one-off tasks) |
| created_at | REAL | Unix timestamp |
### subconscious_log
| Column | Type | Description |
|--------|------|-------------|
| id | TEXT PK | UUID |
| task_id | TEXT | FK to tasks |
| tick_at | REAL | Unix timestamp of the tick |
| decision | TEXT | `in_progress`, `act`, `noop`, `escalate`, `failed`, `cancelled`, `dismissed` |
| result | TEXT | Result text or error message |
| duration_ms | INTEGER | Execution duration |
| created_at | REAL | Unix timestamp |
### subconscious_escalations
| Column | Type | Description |
|--------|------|-------------|
| id | TEXT PK | UUID |
| task_id | TEXT | FK to tasks |
| log_id | TEXT | FK to log entry |
| title | TEXT | Escalation title |
| description | TEXT | What needs approval |
| priority | TEXT | `critical`, `important`, `normal` |
| status | TEXT | `pending`, `approved`, `dismissed` |
| created_at | REAL | Unix timestamp |
| resolved_at | REAL | When approved/dismissed |
---
## RPC Endpoints
All under `openhuman.subconscious_*`:
| Method | Description |
|--------|-------------|
| `subconscious_status` | Get engine status (enabled, ticks, failures) |
| `subconscious_trigger` | Manually trigger a tick (runs in background, returns immediately) |
| `subconscious_tasks_list` | List all tasks |
| `subconscious_tasks_add` | Add a user task |
| `subconscious_tasks_update` | Update task (title, enabled, recurrence) |
| `subconscious_tasks_remove` | Remove a user task (system tasks can only be disabled) |
| `subconscious_log_list` | List activity log entries |
| `subconscious_escalations_list` | List escalations (filterable by status) |
| `subconscious_escalations_approve` | Approve and execute an escalation |
| `subconscious_escalations_dismiss` | Dismiss an escalation |
---
## UI (Intelligence Page → Subconscious Tab)
### Status bar
Shows: task count, total ticks, last tick time, consecutive failures (if > 0).
### Active Tasks
- **System tasks**: displayed as plain text with green dot and "default" badge. No controls.
- **User tasks**: toggle switch (enable/disable) + delete button on hover.
- **Add task**: text input + "Add" button at the bottom.
### Approval Needed
Amber cards for pending escalations. Each shows title, description, priority badge.
- **"Go ahead"**: approve and execute the write action.
- **"Fix in Skills"**: shown for skill-related escalations, navigates to Skills page.
- **"Skip"**: dismiss without executing.
### Activity Log
Chronological list of task evaluations. Each entry shows timestamp, colored dot, and result text. Auto-polls every 2s while any entries are `in_progress`.
### Run Now
Triggers a manual tick. The tick runs in the background — the RPC returns immediately and the UI polls for updates.
@@ -30,9 +30,9 @@ fn default_context_budget() -> u32 {
impl Default for HeartbeatConfig {
fn default() -> Self {
Self {
enabled: false,
interval_minutes: 15,
inference_enabled: false,
enabled: true,
interval_minutes: 5,
inference_enabled: true,
context_budget_tokens: default_context_budget(),
escalation_model: None,
}
+20 -27
View File
@@ -1,6 +1,5 @@
use crate::openhuman::config::HeartbeatConfig;
use crate::openhuman::subconscious::global::get_or_init_engine;
use crate::openhuman::subconscious::types::Decision;
use anyhow::Result;
use std::path::Path;
use tokio::time::{self, Duration};
@@ -40,10 +39,10 @@ impl HeartbeatEngine {
}
);
let mut interval = time::interval(Duration::from_secs(u64::from(interval_mins) * 60));
let sleep_secs = u64::from(interval_mins) * 60;
loop {
interval.tick().await;
time::sleep(Duration::from_secs(sleep_secs)).await;
if self.config.inference_enabled {
// Get the shared global engine (same instance as RPC handlers)
@@ -64,21 +63,12 @@ impl HeartbeatEngine {
};
match engine.tick().await {
Ok(result) => match result.output.decision {
Decision::Noop => {
info!("[heartbeat] tick: noop — {}", result.output.reason);
}
Decision::Act => {
info!(
"[heartbeat] tick: act — {} ({} actions)",
result.output.reason,
result.output.actions.len()
);
}
Decision::Escalate => {
info!("[heartbeat] tick: escalate — {}", result.output.reason);
}
},
Ok(result) => {
info!(
"[heartbeat] tick: executed={} escalated={} duration={}ms",
result.executed, result.escalated, result.duration_ms
);
}
Err(e) => {
warn!("[heartbeat] subconscious tick error: {e}");
}
@@ -124,14 +114,12 @@ impl HeartbeatEngine {
pub async fn ensure_heartbeat_file(workspace_dir: &Path) -> Result<()> {
let path = workspace_dir.join("HEARTBEAT.md");
if !path.exists() {
let default = "# Periodic Tasks\n\
let default = "# Subconscious Instructions\n\
#\n\
# The subconscious loop checks these tasks periodically against\n\
# The subconscious loop evaluates pending tasks periodically against\n\
# your workspace state (memory, skills, email, etc.)\n\
# Add or remove tasks — one per line starting with `- `\n\n\
- Check for new emails that need attention\n\
- Review upcoming deadlines and calendar events\n\
- Monitor connected skills for errors or disconnections\n";
# Tasks are managed in the Subconscious UI — this file provides\n\
# additional context and instructions for task evaluation.\n";
tokio::fs::write(&path, default).await?;
}
Ok(())
@@ -188,9 +176,10 @@ mod tests {
let path = dir.join("HEARTBEAT.md");
assert!(path.exists());
let content = tokio::fs::read_to_string(&path).await.unwrap();
assert!(content.contains("Subconscious Instructions"));
// Instructions only — no task lines
let tasks = HeartbeatEngine::parse_tasks(&content);
assert_eq!(tasks.len(), 3);
assert!(tasks.iter().any(|t| t.contains("email")));
assert_eq!(tasks.len(), 0);
let _ = tokio::fs::remove_dir_all(&dir).await;
}
@@ -214,7 +203,11 @@ mod tests {
#[tokio::test]
async fn run_returns_immediately_when_disabled() {
let engine = HeartbeatEngine::new(HeartbeatConfig::default(), std::env::temp_dir());
let config = HeartbeatConfig {
enabled: false,
..HeartbeatConfig::default()
};
let engine = HeartbeatEngine::new(config, std::env::temp_dir());
let result = engine.run().await;
assert!(result.is_ok());
}
+25 -51
View File
@@ -1,7 +1,7 @@
//! Decision log for tracking what the subconscious has already surfaced.
//! Prevents re-escalating the same state changes across ticks.
use super::types::{Decision, TickOutput};
use super::types::TickDecision;
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
@@ -11,7 +11,7 @@ const RECORD_TTL_SECS: f64 = 24.0 * 60.0 * 60.0;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DecisionRecord {
pub tick_at: f64,
pub decision: Decision,
pub decision: TickDecision,
pub source_doc_ids: Vec<String>,
pub reason: String,
pub acknowledged: bool,
@@ -30,26 +30,22 @@ impl DecisionLog {
}
}
/// Check if any of the given doc IDs have already been surfaced
/// in a non-expired, unacknowledged record.
pub fn was_already_surfaced(&self, doc_ids: &[String]) -> bool {
let now = now_secs();
self.records.iter().any(|r| {
!r.acknowledged
&& r.expires_at > now
&& r.decision != Decision::Noop
&& r.decision != TickDecision::Noop
&& r.source_doc_ids.iter().any(|id| doc_ids.contains(id))
})
}
/// Filter out doc IDs that have already been surfaced.
/// Returns only the new/unseen doc IDs.
pub fn filter_unsurfaced(&self, doc_ids: &[String]) -> Vec<String> {
let surfaced: HashSet<&str> = self
.records
.iter()
.filter(|r| {
!r.acknowledged && r.expires_at > now_secs() && r.decision != Decision::Noop
!r.acknowledged && r.expires_at > now_secs() && r.decision != TickDecision::Noop
})
.flat_map(|r| r.source_doc_ids.iter().map(|s| s.as_str()))
.collect();
@@ -61,19 +57,23 @@ impl DecisionLog {
.collect()
}
/// Record a tick decision.
pub fn record(&mut self, tick_at: f64, output: &TickOutput, source_doc_ids: Vec<String>) {
pub fn record(
&mut self,
tick_at: f64,
decision: TickDecision,
reason: &str,
source_doc_ids: Vec<String>,
) {
self.records.push(DecisionRecord {
tick_at,
decision: output.decision.clone(),
decision,
source_doc_ids,
reason: output.reason.clone(),
reason: reason.to_string(),
acknowledged: false,
expires_at: tick_at + RECORD_TTL_SECS,
});
}
/// Mark all records matching any of the given doc IDs as acknowledged.
pub fn mark_acknowledged(&mut self, doc_ids: &[String]) {
for record in &mut self.records {
if record.source_doc_ids.iter().any(|id| doc_ids.contains(id)) {
@@ -82,29 +82,24 @@ impl DecisionLog {
}
}
/// Remove expired records.
pub fn prune_expired(&mut self) {
let now = now_secs();
self.records.retain(|r| r.expires_at > now);
}
/// Number of active (non-expired) records.
pub fn active_count(&self) -> usize {
let now = now_secs();
self.records.iter().filter(|r| r.expires_at > now).count()
}
/// All records (including expired, for debugging).
pub fn records(&self) -> &[DecisionRecord] {
&self.records
}
/// Serialize to JSON for storage in memory.
pub fn to_json(&self) -> Result<String, String> {
serde_json::to_string(self).map_err(|e| format!("serialize decision log: {e}"))
}
/// Deserialize from JSON.
pub fn from_json(json: &str) -> Result<Self, String> {
serde_json::from_str(json).map_err(|e| format!("deserialize decision log: {e}"))
}
@@ -120,15 +115,6 @@ fn now_secs() -> f64 {
#[cfg(test)]
mod tests {
use super::*;
use crate::openhuman::subconscious::types::{Decision, TickOutput};
fn tick_output(decision: Decision, reason: &str) -> TickOutput {
TickOutput {
decision,
reason: reason.to_string(),
actions: vec![],
}
}
fn now() -> f64 {
now_secs()
@@ -145,7 +131,8 @@ mod tests {
let mut log = DecisionLog::new();
log.record(
now(),
&tick_output(Decision::Escalate, "deadline"),
TickDecision::Escalate,
"deadline",
vec!["doc-1".into()],
);
assert!(log.was_already_surfaced(&["doc-1".into()]));
@@ -155,11 +142,7 @@ mod tests {
#[test]
fn noop_decisions_are_not_surfaced() {
let mut log = DecisionLog::new();
log.record(
now(),
&tick_output(Decision::Noop, "nothing"),
vec!["doc-1".into()],
);
log.record(now(), TickDecision::Noop, "nothing", vec!["doc-1".into()]);
assert!(!log.was_already_surfaced(&["doc-1".into()]));
}
@@ -168,7 +151,8 @@ mod tests {
let mut log = DecisionLog::new();
log.record(
now(),
&tick_output(Decision::Escalate, "deadline"),
TickDecision::Escalate,
"deadline",
vec!["doc-1".into()],
);
log.mark_acknowledged(&["doc-1".into()]);
@@ -181,7 +165,8 @@ mod tests {
let old_time = now() - RECORD_TTL_SECS - 1.0;
log.record(
old_time,
&tick_output(Decision::Escalate, "old"),
TickDecision::Escalate,
"old",
vec!["doc-1".into()],
);
assert!(!log.was_already_surfaced(&["doc-1".into()]));
@@ -193,14 +178,11 @@ mod tests {
let old_time = now() - RECORD_TTL_SECS - 1.0;
log.record(
old_time,
&tick_output(Decision::Escalate, "old"),
TickDecision::Escalate,
"old",
vec!["doc-1".into()],
);
log.record(
now(),
&tick_output(Decision::Act, "new"),
vec!["doc-2".into()],
);
log.record(now(), TickDecision::Act, "new", vec!["doc-2".into()]);
assert_eq!(log.records().len(), 2);
log.prune_expired();
assert_eq!(log.records().len(), 1);
@@ -210,11 +192,7 @@ mod tests {
#[test]
fn filter_unsurfaced_returns_new_docs() {
let mut log = DecisionLog::new();
log.record(
now(),
&tick_output(Decision::Escalate, "seen"),
vec!["doc-1".into()],
);
log.record(now(), TickDecision::Escalate, "seen", vec!["doc-1".into()]);
let unsurfaced = log.filter_unsurfaced(&["doc-1".into(), "doc-2".into(), "doc-3".into()]);
assert_eq!(unsurfaced, vec!["doc-2".to_string(), "doc-3".to_string()]);
}
@@ -222,11 +200,7 @@ mod tests {
#[test]
fn roundtrip_json() {
let mut log = DecisionLog::new();
log.record(
now(),
&tick_output(Decision::Escalate, "test"),
vec!["doc-1".into()],
);
log.record(now(), TickDecision::Escalate, "test", vec!["doc-1".into()]);
let json = log.to_json().unwrap();
let restored = DecisionLog::from_json(&json).unwrap();
assert_eq!(restored.records().len(), 1);
File diff suppressed because it is too large Load Diff
+342
View File
@@ -0,0 +1,342 @@
//! Task execution — dispatches tasks to either the local Ollama model (text-only)
//! or the full agentic loop (tool-required).
//!
//! When agentic-v1 is used for a task that didn't have explicit write intent,
//! it runs in analysis-only mode. If it recommends a write action, execution
//! is paused and an `UnapprovedWrite` result is returned so the engine can
//! create an escalation for user approval.
use super::prompt;
use super::types::{ExecutionResult, SubconsciousTask};
use tracing::{debug, info, warn};
/// Outcome of executing a task — either completed or needs user approval.
pub enum ExecutionOutcome {
/// Task completed (either read-only analysis or approved write).
Completed(ExecutionResult),
/// agentic-v1 recommends a write action on a read-only task.
/// Contains the recommended action description for the escalation.
UnapprovedWrite {
recommendation: String,
duration_ms: u64,
},
}
/// Execute a task. Routes to local model or agentic loop based on whether
/// the task needs external tools.
pub async fn execute_task(
task: &SubconsciousTask,
situation_report: &str,
identity_context: &str,
) -> Result<ExecutionOutcome, String> {
let started = std::time::Instant::now();
let task_has_write_intent = needs_tools(&task.title);
let result = if task_has_write_intent {
// Task explicitly asks for a write action — run with full permissions.
info!(
"[subconscious:executor] write task: {} — agentic loop, full permissions",
task.title
);
execute_with_agent_full(task, situation_report, identity_context)
.await
.map(|output| {
ExecutionOutcome::Completed(ExecutionResult {
output,
used_tools: true,
duration_ms: started.elapsed().as_millis() as u64,
})
})
} else if needs_agent(&task.title) {
// Read-only task but needs deeper reasoning — run analysis-only.
info!(
"[subconscious:executor] read-only task escalated: {} — agentic loop, analysis only",
task.title
);
let output = execute_with_agent_analysis(task, situation_report, identity_context).await?;
let duration_ms = started.elapsed().as_millis() as u64;
if let Some(recommendation) = extract_recommended_action(&output) {
// agentic-v1 wants to take a write action the user didn't ask for.
Ok(ExecutionOutcome::UnapprovedWrite {
recommendation,
duration_ms,
})
} else {
Ok(ExecutionOutcome::Completed(ExecutionResult {
output,
used_tools: false,
duration_ms,
}))
}
} else {
// Simple text-only task — local model handles it.
debug!(
"[subconscious:executor] text task: {} — using local model",
task.title
);
execute_with_local_model(task, situation_report, identity_context)
.await
.map(|output| {
ExecutionOutcome::Completed(ExecutionResult {
output,
used_tools: false,
duration_ms: started.elapsed().as_millis() as u64,
})
})
};
if let Err(ref e) = result {
warn!("[subconscious:executor] task '{}' failed: {e}", task.title);
}
result
}
/// Execute an approved write action — called after user approves an escalation
/// that originated from `UnapprovedWrite`.
pub async fn execute_approved_write(
task: &SubconsciousTask,
situation_report: &str,
identity_context: &str,
) -> Result<ExecutionResult, String> {
let started = std::time::Instant::now();
let output = execute_with_agent_full(task, situation_report, identity_context).await?;
Ok(ExecutionResult {
output,
used_tools: true,
duration_ms: started.elapsed().as_millis() as u64,
})
}
/// Execute a text-only task using the local Ollama model.
async fn execute_with_local_model(
task: &SubconsciousTask,
situation_report: &str,
identity_context: &str,
) -> Result<String, String> {
let config = crate::openhuman::config::Config::load_or_init()
.await
.map_err(|e| format!("config load: {e}"))?;
let prompt_text = prompt::build_text_execution_prompt(task, situation_report, identity_context);
let messages = vec![
crate::openhuman::local_ai::ops::LocalAiChatMessage {
role: "system".to_string(),
content: prompt_text,
},
crate::openhuman::local_ai::ops::LocalAiChatMessage {
role: "user".to_string(),
content: "Execute the task now.".to_string(),
},
];
let outcome = crate::openhuman::local_ai::ops::local_ai_chat(&config, messages, None)
.await
.map_err(|e| format!("local model: {e}"))?;
Ok(outcome.value)
}
/// Execute with agentic-v1 at full permissions (write-intent tasks or approved writes).
///
/// Retries up to 3 times with exponential backoff (2s, 4s, 8s) on 429 rate-limit
/// errors from the agentic-v1 cloud model.
async fn execute_with_agent_full(
task: &SubconsciousTask,
situation_report: &str,
identity_context: &str,
) -> Result<String, String> {
let mut config = crate::openhuman::config::Config::load_or_init()
.await
.map_err(|e| format!("config load: {e}"))?;
let prompt_text = prompt::build_tool_execution_prompt(task, situation_report, identity_context);
agent_chat_with_retry(&mut config, &prompt_text).await
}
/// Execute with agentic-v1 in analysis-only mode (read-only tasks).
///
/// The prompt instructs the model to analyze but not execute write actions.
async fn execute_with_agent_analysis(
task: &SubconsciousTask,
situation_report: &str,
identity_context: &str,
) -> Result<String, String> {
let mut config = crate::openhuman::config::Config::load_or_init()
.await
.map_err(|e| format!("config load: {e}"))?;
let prompt_text = prompt::build_analysis_only_prompt(task, situation_report, identity_context);
agent_chat_with_retry(&mut config, &prompt_text).await
}
/// Call agent_chat with rate-limit retry (429 only, up to 3 attempts).
async fn agent_chat_with_retry(
config: &mut crate::openhuman::config::Config,
prompt: &str,
) -> Result<String, String> {
const MAX_RETRIES: u32 = 3;
let mut attempt = 0;
loop {
let result =
crate::openhuman::local_ai::ops::agent_chat(config, prompt, None, Some(0.3)).await;
match result {
Ok(outcome) => return Ok(outcome.value),
Err(e) => {
let is_rate_limit = e.contains("429") || e.to_lowercase().contains("rate limit");
attempt += 1;
if is_rate_limit && attempt < MAX_RETRIES {
let backoff_secs = 2u64 << (attempt - 1); // 2, 4, 8
warn!(
"[subconscious:executor] rate-limited (attempt {}/{}), retrying in {}s: {}",
attempt, MAX_RETRIES, backoff_secs, e
);
tokio::time::sleep(std::time::Duration::from_secs(backoff_secs)).await;
continue;
}
return Err(format!("agent execution: {e}"));
}
}
}
}
/// Check if the analysis output contains a recommended write action.
/// Returns the recommendation text if found.
fn extract_recommended_action(output: &str) -> Option<String> {
// Look for "RECOMMENDED ACTION:" marker in the output
for line_idx in output.lines().enumerate().filter_map(|(i, l)| {
if l.trim().starts_with("RECOMMENDED ACTION:") {
Some(i)
} else {
None
}
}) {
let recommendation: String = output
.lines()
.skip(line_idx)
.collect::<Vec<_>>()
.join("\n")
.trim()
.to_string();
if !recommendation.is_empty() {
return Some(recommendation);
}
}
None
}
/// Heuristic: does this task need the agentic loop (deeper reasoning, tools)?
///
/// Tasks escalated by the local model that involve complex analysis
/// (multi-step reasoning, cross-referencing sources) benefit from agentic-v1
/// even without write actions.
fn needs_agent(title: &str) -> bool {
let lower = title.to_lowercase();
let agent_keywords = [
"compare",
"cross-reference",
"correlate",
"investigate",
"deep dive",
"research",
"audit",
"trace",
"debug",
"diagnose",
];
agent_keywords.iter().any(|kw| lower.contains(kw))
}
/// Heuristic: does this task description imply needing external tools?
///
/// Tasks with action verbs (send, create, post, delete, move, publish, schedule)
/// need the agentic loop. Tasks with passive verbs (summarize, check, monitor,
/// review, analyze, extract, classify) can be handled by local model.
pub fn needs_tools(title: &str) -> bool {
let lower = title.to_lowercase();
let tool_keywords = [
"send",
"post",
"create",
"delete",
"remove",
"move",
"publish",
"schedule",
"forward",
"reply",
"draft and send",
"upload",
"download",
"notify on",
"alert on",
"message",
"write to",
"update on",
"sync to",
];
tool_keywords.iter().any(|kw| lower.contains(kw))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn needs_tools_detects_action_verbs() {
assert!(needs_tools("Send email digest to Telegram"));
assert!(needs_tools("Post weekly standup to Slack"));
assert!(needs_tools("Create a summary in Notion"));
assert!(needs_tools("Delete old calendar events"));
assert!(needs_tools("Forward urgent emails to team"));
assert!(needs_tools("Schedule a meeting for tomorrow"));
}
#[test]
fn needs_tools_rejects_passive_verbs() {
assert!(!needs_tools("Summarize unread emails"));
assert!(!needs_tools("Check skills runtime health"));
assert!(!needs_tools("Monitor Ollama status"));
assert!(!needs_tools("Review upcoming deadlines"));
assert!(!needs_tools("Analyze email patterns"));
assert!(!needs_tools("Extract key points from Notion pages"));
assert!(!needs_tools("Classify email priority"));
}
#[test]
fn needs_tools_case_insensitive() {
assert!(needs_tools("SEND a message to Slack"));
assert!(needs_tools("Send A Message To Slack"));
}
#[test]
fn needs_agent_detects_complex_tasks() {
assert!(needs_agent("Compare Q1 and Q2 revenue data"));
assert!(needs_agent("Investigate why notifications stopped"));
assert!(needs_agent("Audit all active skill connections"));
assert!(!needs_agent("Check emails"));
assert!(!needs_agent("Summarize today's events"));
}
#[test]
fn extract_recommended_action_finds_marker() {
let output = "Analysis complete. Found 3 urgent emails.\n\nRECOMMENDED ACTION: Forward the 3 urgent emails to #team-alerts on Slack.";
let action = extract_recommended_action(output);
assert!(action.is_some());
assert!(action.unwrap().contains("Forward"));
}
#[test]
fn extract_recommended_action_returns_none_when_absent() {
let output = "All skills are healthy. No issues found.";
assert!(extract_recommended_action(output).is_none());
}
}
+199 -253
View File
@@ -1,279 +1,225 @@
#[cfg(test)]
mod tests {
use std::sync::Arc;
use serde_json::json;
use tempfile::TempDir;
use crate::openhuman::memory::embeddings::NoopEmbedding;
use crate::openhuman::memory::{
MemoryIngestionConfig, MemoryIngestionRequest, NamespaceDocumentInput, UnifiedMemory,
};
use crate::openhuman::subconscious::decision_log::DecisionLog;
use crate::openhuman::subconscious::situation_report::build_situation_report;
use crate::openhuman::subconscious::types::Decision;
use crate::openhuman::subconscious::store;
use crate::openhuman::subconscious::types::{
EscalationPriority, EscalationStatus, TaskRecurrence, TaskSource, TickDecision,
};
/// Find the largest byte index ≤ `max_bytes` that is a valid char boundary.
fn truncate_at_char_boundary(s: &str, max_bytes: usize) -> usize {
let mut end = s.len().min(max_bytes);
while end > 0 && !s.is_char_boundary(end) {
end -= 1;
}
end
#[test]
fn sqlite_task_lifecycle_one_off() {
let dir = tempfile::tempdir().unwrap();
store::with_connection(dir.path(), |conn| {
// Add a one-off task
let task = store::add_task(
conn,
"Remind about meeting",
TaskSource::User,
TaskRecurrence::Once,
)?;
assert!(!task.completed);
assert_eq!(task.recurrence, TaskRecurrence::Once);
// Should be due immediately
let due = store::due_tasks(conn, 9999999999.0)?;
assert_eq!(due.len(), 1);
// Execute and complete
store::add_log_entry(
conn,
&task.id,
1000.0,
"act",
Some("Reminded user"),
Some(50),
)?;
store::mark_task_completed(conn, &task.id)?;
// Should no longer be due
let due = store::due_tasks(conn, 9999999999.0)?;
assert_eq!(due.len(), 0);
// Task still exists but completed
let t = store::get_task(conn, &task.id)?;
assert!(t.completed);
Ok(())
})
.unwrap();
}
fn fixture(path: &str) -> String {
let base = std::path::Path::new(env!("CARGO_MANIFEST_DIR"));
std::fs::read_to_string(
base.join("tests")
.join("fixtures")
.join("subconscious")
.join(path),
)
.expect("fixture should load")
#[test]
fn sqlite_task_lifecycle_recurrent() {
let dir = tempfile::tempdir().unwrap();
store::with_connection(dir.path(), |conn| {
let task = store::add_task(
conn,
"Check email",
TaskSource::User,
TaskRecurrence::Cron("0 8 * * *".into()),
)?;
// Execute and set next run
let now = 1000.0;
let next = 2000.0;
store::add_log_entry(
conn,
&task.id,
now,
"act",
Some("Checked 3 emails"),
Some(200),
)?;
store::update_task_run_times(conn, &task.id, now, Some(next))?;
// Not due yet (before next_run_at)
let due = store::due_tasks(conn, 1500.0)?;
assert_eq!(due.len(), 0);
// Due after next_run_at
let due = store::due_tasks(conn, 2500.0)?;
assert_eq!(due.len(), 1);
// Task should NOT be completed
let t = store::get_task(conn, &task.id)?;
assert!(!t.completed);
Ok(())
})
.unwrap();
}
fn ci_safe_config() -> MemoryIngestionConfig {
MemoryIngestionConfig {
model_name: "__test_no_model__".to_string(),
..MemoryIngestionConfig::default()
}
#[test]
fn escalation_approve_dismiss_flow() {
let dir = tempfile::tempdir().unwrap();
store::with_connection(dir.path(), |conn| {
let task = store::add_task(
conn,
"Review deadline",
TaskSource::User,
TaskRecurrence::Once,
)?;
// Create escalation
let esc = store::add_escalation(
conn,
&task.id,
None,
"Deadline conflict",
"Two deadlines on the same day",
&EscalationPriority::Important,
)?;
assert_eq!(esc.status, EscalationStatus::Pending);
assert_eq!(store::pending_escalation_count(conn)?, 1);
// Approve
store::resolve_escalation(conn, &esc.id, &EscalationStatus::Approved)?;
let resolved = store::get_escalation(conn, &esc.id)?;
assert_eq!(resolved.status, EscalationStatus::Approved);
assert!(resolved.resolved_at.is_some());
assert_eq!(store::pending_escalation_count(conn)?, 0);
// Create another and dismiss
let esc2 = store::add_escalation(
conn,
&task.id,
None,
"Budget warning",
"Monthly spend at 90%",
&EscalationPriority::Normal,
)?;
store::resolve_escalation(conn, &esc2.id, &EscalationStatus::Dismissed)?;
let dismissed = store::get_escalation(conn, &esc2.id)?;
assert_eq!(dismissed.status, EscalationStatus::Dismissed);
Ok(())
})
.unwrap();
}
async fn ingest(
memory: &UnifiedMemory,
namespace: &str,
key: &str,
title: &str,
content: &str,
) -> String {
let result = memory
.ingest_document(MemoryIngestionRequest {
document: NamespaceDocumentInput {
namespace: namespace.to_string(),
key: key.to_string(),
title: title.to_string(),
content: content.to_string(),
source_type: "test".to_string(),
priority: "high".to_string(),
tags: Vec::new(),
metadata: json!({}),
category: "core".to_string(),
session_id: None,
document_id: None,
},
config: ci_safe_config(),
})
.await
.unwrap();
result.document_id
#[test]
fn execution_log_tracks_history() {
let dir = tempfile::tempdir().unwrap();
store::with_connection(dir.path(), |conn| {
let task = store::add_task(
conn,
"Check health",
TaskSource::System,
TaskRecurrence::Pending,
)?;
store::add_log_entry(conn, &task.id, 1000.0, "noop", Some("All healthy"), None)?;
store::add_log_entry(
conn,
&task.id,
2000.0,
"act",
Some("Restarted skill"),
Some(500),
)?;
store::add_log_entry(conn, &task.id, 3000.0, "noop", Some("All healthy"), None)?;
let entries = store::list_log_entries(conn, Some(&task.id), 10)?;
assert_eq!(entries.len(), 3);
// Most recent first
assert_eq!(entries[0].tick_at, 3000.0);
assert_eq!(entries[1].decision, "act");
// Global log
let all = store::list_log_entries(conn, None, 2)?;
assert_eq!(all.len(), 2); // limited to 2
Ok(())
})
.unwrap();
}
/// Full two-tick integration test:
/// 1. Ingest tick1 data → build report → verify it contains the data
/// 2. Record a decision in the log
/// 3. Ingest tick2 data → build report → verify delta-only (not old data)
/// 4. Verify decision log deduplication
#[tokio::test]
async fn two_tick_lifecycle() {
let tmp = TempDir::new().unwrap();
let workspace = tmp.path();
let memory = UnifiedMemory::new(workspace, Arc::new(NoopEmbedding), None).unwrap();
let client =
crate::openhuman::memory::MemoryClient::from_workspace_dir(workspace.to_path_buf())
.unwrap();
// Write HEARTBEAT.md
std::fs::write(workspace.join("HEARTBEAT.md"), fixture("heartbeat.md")).unwrap();
// ============================================================
// TICK 1: Ingest initial data
// ============================================================
let tick1_time = std::time::SystemTime::now()
#[test]
fn decision_log_dedup_still_works() {
let mut log = DecisionLog::new();
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs_f64();
let gmail_doc_id = ingest(
&memory,
"skill-gmail",
"tick1-deadline-email",
"API contract deadline reminder",
&fixture("tick1_gmail.txt"),
)
.await;
let _notion_doc_id = ingest(
&memory,
"skill-notion",
"tick1-tracker",
"Q1 Delivery Tracker",
&fixture("tick1_notion.txt"),
)
.await;
// Build situation report for tick 1 (cold start: last_tick_at = 0)
let report1 = build_situation_report(
Some(&client),
workspace,
0.0, // cold start
40_000,
)
.await;
println!("=== TICK 1 REPORT ===");
println!("{}", &report1[..truncate_at_char_boundary(&report1, 2000)]);
println!("=====================\n");
// Verify tick 1 report contains ingested data
assert!(
report1.contains("Memory Documents"),
"Report should have memory docs section"
);
assert!(
report1.contains("skill-gmail") || report1.contains("deadline"),
"Report should mention gmail data"
);
assert!(
report1.contains("Pending Tasks"),
"Report should have tasks section"
);
assert!(
report1.contains("Check for deadline changes"),
"Report should include HEARTBEAT.md tasks"
log.record(
now,
TickDecision::Escalate,
"deadline email",
vec!["doc-1".into()],
);
// Simulate tick 1 decision: escalate the deadline
let mut decision_log = DecisionLog::new();
let tick1_output = crate::openhuman::subconscious::types::TickOutput {
decision: Decision::Escalate,
reason: "Deadline reminder for API contract (April 3)".to_string(),
actions: vec![],
};
decision_log.record(tick1_time, &tick1_output, vec![gmail_doc_id.clone()]);
// doc-1 should be filtered as already surfaced
let unsurfaced = log.filter_unsurfaced(&["doc-1".into(), "doc-2".into()]);
assert!(!unsurfaced.contains(&"doc-1".to_string()));
assert!(unsurfaced.contains(&"doc-2".to_string()));
println!(
"Decision log after tick 1: {} active records",
decision_log.active_count()
);
assert_eq!(decision_log.active_count(), 1);
assert!(decision_log.was_already_surfaced(&[gmail_doc_id.clone()]));
assert!(!decision_log.was_already_surfaced(&["nonexistent".to_string()]));
// Acknowledge doc-1
log.mark_acknowledged(&["doc-1".into()]);
assert!(!log.was_already_surfaced(&["doc-1".into()]));
}
// ============================================================
// TICK 2: Ingest new data (state change)
// ============================================================
let tick2_time = tick1_time + 1.0; // 1 second later (simulated)
#[test]
fn seed_then_query_tasks() {
let dir = tempfile::tempdir().unwrap();
let gmail2_doc_id = ingest(
&memory,
"skill-gmail",
"tick2-deadline-moved",
"URGENT: API contract deadline moved to tomorrow",
&fixture("tick2_gmail.txt"),
)
.await;
store::with_connection(dir.path(), |conn| {
let count = store::seed_default_tasks(conn)?;
assert_eq!(count, 3);
let notion2_doc_id = ingest(
&memory,
"skill-notion",
"tick2-tracker-updated",
"Q1 Delivery Tracker (updated)",
&fixture("tick2_notion.txt"),
)
.await;
let tasks = store::list_tasks(conn, true)?;
assert_eq!(tasks.len(), 3);
assert!(tasks.iter().all(|t| t.source == TaskSource::System));
assert!(tasks
.iter()
.all(|t| t.recurrence == TaskRecurrence::Pending));
// Build situation report for tick 2 (delta since tick 1)
let report2 = build_situation_report(
Some(&client),
workspace,
tick1_time, // delta since tick 1
40_000,
)
.await;
// All should be due (no next_run_at set)
let due = store::due_tasks(conn, 9999999999.0)?;
assert_eq!(due.len(), 3);
println!("=== TICK 2 REPORT ===");
println!("{}", &report2[..truncate_at_char_boundary(&report2, 2000)]);
println!("=====================\n");
// Verify tick 2 report contains NEW data
assert!(
report2.contains("new/updated"),
"Report should show new/updated docs"
);
// Verify deduplication: old gmail doc should be filtered
let all_new_doc_ids = vec![
gmail_doc_id.clone(),
gmail2_doc_id.clone(),
notion2_doc_id.clone(),
];
let unsurfaced = decision_log.filter_unsurfaced(&all_new_doc_ids);
println!("Unsurfaced doc IDs: {:?}", unsurfaced);
assert!(
!unsurfaced.contains(&gmail_doc_id),
"Old deadline email should be filtered out (already surfaced)"
);
assert!(
unsurfaced.contains(&gmail2_doc_id),
"New deadline-moved email should NOT be filtered"
);
assert!(
unsurfaced.contains(&notion2_doc_id),
"Updated notion tracker should NOT be filtered"
);
// Record tick 2 decision
let tick2_output = crate::openhuman::subconscious::types::TickOutput {
decision: Decision::Escalate,
reason: "Deadline moved to tomorrow — urgent".to_string(),
actions: vec![],
};
decision_log.record(tick2_time, &tick2_output, vec![gmail2_doc_id.clone()]);
println!(
"Decision log after tick 2: {} active records",
decision_log.active_count()
);
assert_eq!(decision_log.active_count(), 2);
// ============================================================
// TICK 3: No new data
// ============================================================
let report3 = build_situation_report(
Some(&client),
workspace,
tick2_time, // delta since tick 2
40_000,
)
.await;
println!("=== TICK 3 REPORT ===");
println!("{}", &report3[..truncate_at_char_boundary(&report3, 2000)]);
println!("=====================\n");
// Tick 3 should show no changes
let has_changes = !report3.contains("No changes since last tick");
// Note: on cold data with fixed timestamps, all docs may appear
// "old" relative to tick2_time. The key test is that the decision
// log correctly filters previously surfaced items.
println!("Tick 3 has changes: {}", has_changes);
// Verify JSON roundtrip of decision log
let json = decision_log.to_json().unwrap();
let restored = DecisionLog::from_json(&json).unwrap();
assert_eq!(restored.active_count(), 2);
assert!(restored.was_already_surfaced(&[gmail_doc_id.clone()]));
assert!(restored.was_already_surfaced(&[gmail2_doc_id.clone()]));
// Verify acknowledgment
decision_log.mark_acknowledged(&[gmail_doc_id.clone()]);
assert!(
!decision_log.was_already_surfaced(&[gmail_doc_id]),
"Acknowledged docs should no longer be surfaced"
);
println!("=== ALL TESTS PASSED ===");
Ok(())
})
.unwrap();
}
}
+9 -2
View File
@@ -1,11 +1,15 @@
pub mod decision_log;
pub mod engine;
pub mod executor;
pub mod global;
pub mod prompt;
mod schemas;
pub mod situation_report;
pub mod store;
pub mod types;
// Keep decision_log for potential future dedup queries against the log table.
pub mod decision_log;
#[cfg(test)]
mod integration_test;
@@ -14,4 +18,7 @@ pub use schemas::{
all_controller_schemas as all_subconscious_controller_schemas,
all_registered_controllers as all_subconscious_registered_controllers,
};
pub use types::{Decision, SubconsciousStatus, TickOutput, TickResult};
pub use types::{
Escalation, EscalationStatus, SubconsciousLogEntry, SubconsciousStatus, SubconsciousTask,
TaskRecurrence, TaskSource, TickDecision, TickResult,
};
+229 -56
View File
@@ -1,24 +1,37 @@
//! System prompt for the subconscious local-model evaluation.
//! Prompt builders for the subconscious evaluation and execution phases.
//!
//! Injects OpenClaw identity context (SOUL.md, USER.md) so the local model
//! reasons as the agent, not a generic evaluator.
/// Build the task-driven system prompt for the subconscious tick.
///
/// The local model evaluates each HEARTBEAT.md task against the current
/// state and decides which tasks have actionable items right now.
pub fn build_subconscious_prompt(tasks: &[String], situation_report: &str) -> String {
use super::types::SubconsciousTask;
use std::path::Path;
const IDENTITY_EXCERPT_CHARS: usize = 2000;
// ── Evaluation prompt ────────────────────────────────────────────────────────
/// Build the per-tick evaluation prompt. The local model evaluates each due
/// task against the situation report and returns a per-task decision.
pub fn build_evaluation_prompt(
tasks: &[SubconsciousTask],
situation_report: &str,
identity_context: &str,
) -> String {
let task_list = tasks
.iter()
.enumerate()
.map(|(i, t)| format!("{}. {}", i + 1, t))
.map(|t| format!("- [{}] {}", t.id, t.title))
.collect::<Vec<_>>()
.join("\n");
format!(
r#"# Subconscious Loop — Task Evaluation
r#"{identity_context}
You are the background awareness layer of OpenHuman. You run periodically
to check a list of user-defined tasks against the current workspace state.
# Subconscious Loop — Task Evaluation
## Your tasks to check
You are the background awareness layer. You run periodically to evaluate
user-defined tasks against the current workspace state.
## Due tasks
{task_list}
@@ -28,73 +41,233 @@ to check a list of user-defined tasks against the current workspace state.
## Your job
For each task above, check if the current state contains anything relevant.
If a task has something actionable, include it in the actions list.
If no tasks have anything actionable, return noop.
For each task, check if the current state has anything relevant. Decide:
- **noop**: Nothing actionable for this task right now.
- **act**: The task should be executed now (state has relevant data).
- **escalate**: The task needs user approval before acting (ambiguous, risky, or irreversible).
## Output format (strict JSON, no other text)
{{
"decision": "noop" | "act" | "escalate",
"reason": "one sentence summary of what you found",
"actions": [
{{
"type": "notify" | "store_memory" | "escalate_to_agent",
"description": "what was found and what to do about it",
"priority": "low" | "medium" | "high",
"task": "which task this relates to"
}}
"evaluations": [
{{"task_id": "<id>", "decision": "noop|act|escalate", "reason": "one sentence"}}
]
}}
## Decision rules
- **noop**: None of the tasks have anything actionable in the current state.
- **act**: One or more tasks have findings that can be summarized as a notification or stored as a memory note.
- **escalate**: A task finding requires complex reasoning or multi-step action that the full agent should handle (e.g. drafting a response, reprioritizing work, multi-tool operations).
## Examples
Task: "Check email for urgent items"
State shows: new email about deadline moved to tomorrow
→ act, notify with high priority
Task: "Monitor skills runtime health"
State shows: no skill data available
→ noop for this task
Task: "Check for deadline changes"
State shows: project tracker updated, 3 deadlines shifted, ownership changed
→ escalate (too complex for simple notification)
"#
)
}
// ── Execution prompts ────────────────────────────────────────────────────────
/// Build the prompt for executing a text-only task via local Ollama model.
/// Used for tasks that don't need tools (summarize, extract, classify, etc.)
pub fn build_text_execution_prompt(
task: &SubconsciousTask,
situation_report: &str,
identity_context: &str,
) -> String {
format!(
r#"{identity_context}
# Task Execution
Execute the following task based on the current state. Respond with the result only.
## Task
{task_title}
## Current state
{situation_report}
Do the task now. Return only the result — no explanations or meta-commentary."#,
task_title = task.title
)
}
/// Build the prompt for executing a tool-required task via the full agentic loop.
/// Used for tasks that need side effects (send message, create doc, etc.)
pub fn build_tool_execution_prompt(
task: &SubconsciousTask,
situation_report: &str,
identity_context: &str,
) -> String {
format!(
r#"{identity_context}
# Background Task Execution
You are executing a user-defined background task. Use your available tools to complete it.
## Task
{task_title}
## Current state
{situation_report}
Execute this task using the appropriate tools. Complete the task fully — don't just describe what to do."#,
task_title = task.title
)
}
/// Build a read-only analysis prompt for agentic-v1. Used when a read-only task
/// is escalated — the agent should analyze and recommend but NOT execute writes.
pub fn build_analysis_only_prompt(
task: &SubconsciousTask,
situation_report: &str,
identity_context: &str,
) -> String {
format!(
r#"{identity_context}
# Background Task — Analysis Only
You are analyzing a background task. You may use read-only tools to gather information,
but you MUST NOT execute any write actions (send, post, create, delete, forward, reply, update, publish).
If you determine that a write action is needed, describe exactly what you would do in your response
but do not execute it. Start your recommendation with "RECOMMENDED ACTION:" on its own line.
## Task
{task_title}
## Current state
{situation_report}
Analyze the situation and report your findings. If action is needed, describe it clearly but do NOT execute."#,
task_title = task.title
)
}
// ── Identity loading ─────────────────────────────────────────────────────────
/// Load identity context from SOUL.md and USER.md in the prompts directory.
/// Returns a formatted string to prepend to prompts.
pub fn load_identity_context(workspace_dir: &Path) -> String {
let prompts_dir = resolve_prompts_dir(workspace_dir);
let mut ctx = String::new();
if let Some(ref dir) = prompts_dir {
if let Some(soul) = load_file_excerpt(dir, "SOUL.md") {
ctx.push_str(&soul);
ctx.push_str("\n\n");
}
if let Some(user) = load_file_excerpt(dir, "USER.md") {
ctx.push_str("## User Context\n\n");
ctx.push_str(&user);
ctx.push_str("\n\n");
}
}
if ctx.is_empty() {
"You are OpenHuman, an AI assistant for productivity and collaboration.".to_string()
} else {
ctx
}
}
fn resolve_prompts_dir(workspace_dir: &Path) -> Option<std::path::PathBuf> {
// Check workspace AI dir
let workspace_ai = workspace_dir.join("ai");
if workspace_ai.is_dir() {
return Some(workspace_ai);
}
// Try CARGO_MANIFEST_DIR (dev builds)
if let Some(dir) = option_env!("CARGO_MANIFEST_DIR").map(std::path::PathBuf::from) {
let candidate = dir
.join("src")
.join("openhuman")
.join("agent")
.join("prompts");
if candidate.is_dir() {
return Some(candidate);
}
}
// Walk up from cwd
if let Ok(cwd) = std::env::current_dir() {
return crate::openhuman::dev_paths::repo_ai_prompts_dir(&cwd);
}
None
}
fn load_file_excerpt(dir: &Path, filename: &str) -> Option<String> {
let content = std::fs::read_to_string(dir.join(filename)).ok()?;
let trimmed = content.trim();
if trimmed.is_empty() {
return None;
}
if trimmed.chars().count() > IDENTITY_EXCERPT_CHARS {
let truncated: String = trimmed.chars().take(IDENTITY_EXCERPT_CHARS).collect();
Some(format!("{truncated}\n[... truncated]"))
} else {
Some(trimmed.to_string())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::openhuman::subconscious::types::{TaskRecurrence, TaskSource};
fn test_task(id: &str, title: &str) -> SubconsciousTask {
SubconsciousTask {
id: id.to_string(),
title: title.to_string(),
source: TaskSource::User,
recurrence: TaskRecurrence::Once,
enabled: true,
last_run_at: None,
next_run_at: None,
completed: false,
created_at: 0.0,
}
}
#[test]
fn prompt_includes_tasks_and_report() {
let tasks = vec!["Check email".to_string(), "Review calendar".to_string()];
let prompt = build_subconscious_prompt(&tasks, "## Test\n\nSome data.");
assert!(prompt.contains("1. Check email"));
assert!(prompt.contains("2. Review calendar"));
assert!(prompt.contains("## Test"));
fn evaluation_prompt_includes_tasks_and_report() {
let tasks = vec![
test_task("t1", "Check email"),
test_task("t2", "Review calendar"),
];
let prompt = build_evaluation_prompt(&tasks, "## State\nSome data.", "Identity here");
assert!(prompt.contains("[t1] Check email"));
assert!(prompt.contains("[t2] Review calendar"));
assert!(prompt.contains("Some data."));
assert!(prompt.contains("Identity here"));
}
#[test]
fn prompt_includes_json_schema() {
let prompt = build_subconscious_prompt(&["Task".into()], "");
fn evaluation_prompt_includes_decision_schema() {
let tasks = vec![test_task("t1", "Task")];
let prompt = build_evaluation_prompt(&tasks, "", "");
assert!(prompt.contains("noop"));
assert!(prompt.contains("act"));
assert!(prompt.contains("escalate"));
assert!(prompt.contains("escalate_to_agent"));
assert!(prompt.contains("evaluations"));
assert!(prompt.contains("task_id"));
}
#[test]
fn prompt_includes_task_field_in_actions() {
let prompt = build_subconscious_prompt(&["Task".into()], "");
assert!(prompt.contains("\"task\""));
fn text_execution_prompt_includes_task_title() {
let task = test_task("t1", "Summarize urgent emails");
let prompt = build_text_execution_prompt(&task, "3 new emails", "Identity");
assert!(prompt.contains("Summarize urgent emails"));
assert!(prompt.contains("3 new emails"));
}
#[test]
fn tool_execution_prompt_includes_tool_instructions() {
let task = test_task("t1", "Send digest to Telegram");
let prompt = build_tool_execution_prompt(&task, "Email data here", "Identity");
assert!(prompt.contains("Send digest to Telegram"));
assert!(prompt.contains("tools"));
}
#[test]
fn identity_context_loads_or_falls_back() {
let ctx = load_identity_context(std::path::Path::new("/nonexistent"));
assert!(ctx.contains("OpenHuman"));
}
}
+402 -102
View File
@@ -1,12 +1,27 @@
//! RPC endpoints for the subconscious task system.
use serde_json::{Map, Value};
use super::global::get_or_init_engine;
use super::store;
use super::types::{EscalationStatus, TaskPatch, TaskRecurrence, TaskSource};
use crate::core::all::{ControllerFuture, RegisteredController};
use crate::core::{ControllerSchema, FieldSchema, TypeSchema};
use crate::rpc::RpcOutcome;
pub fn all_controller_schemas() -> Vec<ControllerSchema> {
vec![schemas("status"), schemas("trigger"), schemas("actions")]
vec![
schemas("status"),
schemas("trigger"),
schemas("tasks_list"),
schemas("tasks_add"),
schemas("tasks_update"),
schemas("tasks_remove"),
schemas("log_list"),
schemas("escalations_list"),
schemas("escalations_approve"),
schemas("escalations_dismiss"),
]
}
pub fn all_registered_controllers() -> Vec<RegisteredController> {
@@ -20,8 +35,36 @@ pub fn all_registered_controllers() -> Vec<RegisteredController> {
handler: handle_trigger,
},
RegisteredController {
schema: schemas("actions"),
handler: handle_actions,
schema: schemas("tasks_list"),
handler: handle_tasks_list,
},
RegisteredController {
schema: schemas("tasks_add"),
handler: handle_tasks_add,
},
RegisteredController {
schema: schemas("tasks_update"),
handler: handle_tasks_update,
},
RegisteredController {
schema: schemas("tasks_remove"),
handler: handle_tasks_remove,
},
RegisteredController {
schema: schemas("log_list"),
handler: handle_log_list,
},
RegisteredController {
schema: schemas("escalations_list"),
handler: handle_escalations_list,
},
RegisteredController {
schema: schemas("escalations_approve"),
handler: handle_escalations_approve,
},
RegisteredController {
schema: schemas("escalations_dismiss"),
handler: handle_escalations_dismiss,
},
]
}
@@ -31,72 +74,165 @@ pub fn schemas(function: &str) -> ControllerSchema {
"status" => ControllerSchema {
namespace: "subconscious",
function: "status",
description: "Get the current subconscious loop status.",
description: "Get the current subconscious engine status.",
inputs: vec![],
outputs: vec![FieldSchema {
name: "result",
ty: TypeSchema::Json,
comment: "Loop status including last tick, decision counts.",
required: true,
}],
outputs: vec![field("result", TypeSchema::Json, "Engine status.")],
},
"trigger" => ControllerSchema {
namespace: "subconscious",
function: "trigger",
description: "Manually trigger a subconscious tick.",
inputs: vec![],
outputs: vec![FieldSchema {
name: "result",
ty: TypeSchema::Json,
comment: "Tick result with decision, reason, and actions.",
required: true,
}],
outputs: vec![field("result", TypeSchema::Json, "Tick result.")],
},
"actions" => ControllerSchema {
"tasks_list" => ControllerSchema {
namespace: "subconscious",
function: "actions",
description: "List stored subconscious actions/notifications.",
inputs: vec![FieldSchema {
name: "limit",
ty: TypeSchema::Option(Box::new(TypeSchema::U64)),
comment: "Maximum number of actions to return (default: 20).",
required: false,
}],
outputs: vec![FieldSchema {
name: "actions",
ty: TypeSchema::Json,
comment: "Array of stored action entries with timestamps.",
required: true,
}],
function: "tasks_list",
description: "List all subconscious tasks.",
inputs: vec![field_opt(
"enabled_only",
TypeSchema::Bool,
"Filter to enabled tasks only.",
)],
outputs: vec![field("tasks", TypeSchema::Json, "Array of tasks.")],
},
"tasks_add" => ControllerSchema {
namespace: "subconscious",
function: "tasks_add",
description: "Add a new task. The agent classifies it as one-off or recurrent.",
inputs: vec![
field_req(
"title",
TypeSchema::String,
"Natural language task description.",
),
field_opt(
"source",
TypeSchema::String,
"Task source: 'user' (default) or 'system'.",
),
],
outputs: vec![field("task", TypeSchema::Json, "The created task.")],
},
"tasks_update" => ControllerSchema {
namespace: "subconscious",
function: "tasks_update",
description: "Update a task.",
inputs: vec![
field_req("task_id", TypeSchema::String, "Task ID to update."),
field_opt("title", TypeSchema::String, "New title."),
field_opt(
"recurrence",
TypeSchema::String,
"New recurrence: 'once' | 'cron:<expr>' | 'pending'.",
),
field_opt("enabled", TypeSchema::Bool, "Enable or disable."),
],
outputs: vec![field("result", TypeSchema::Json, "Update confirmation.")],
},
"tasks_remove" => ControllerSchema {
namespace: "subconscious",
function: "tasks_remove",
description: "Remove a task.",
inputs: vec![field_req(
"task_id",
TypeSchema::String,
"Task ID to remove.",
)],
outputs: vec![field("result", TypeSchema::Json, "Removal confirmation.")],
},
"log_list" => ControllerSchema {
namespace: "subconscious",
function: "log_list",
description: "List execution log entries.",
inputs: vec![
field_opt("task_id", TypeSchema::String, "Filter by task ID."),
field_opt("limit", TypeSchema::U64, "Max entries (default 50)."),
],
outputs: vec![field("entries", TypeSchema::Json, "Log entries.")],
},
"escalations_list" => ControllerSchema {
namespace: "subconscious",
function: "escalations_list",
description: "List escalations.",
inputs: vec![field_opt(
"status",
TypeSchema::String,
"Filter: 'pending' | 'approved' | 'dismissed'.",
)],
outputs: vec![field(
"escalations",
TypeSchema::Json,
"Escalation records.",
)],
},
"escalations_approve" => ControllerSchema {
namespace: "subconscious",
function: "escalations_approve",
description: "Approve an escalation — execute the task.",
inputs: vec![field_req(
"escalation_id",
TypeSchema::String,
"Escalation ID.",
)],
outputs: vec![field("result", TypeSchema::Json, "Approval confirmation.")],
},
"escalations_dismiss" => ControllerSchema {
namespace: "subconscious",
function: "escalations_dismiss",
description: "Dismiss an escalation — don't execute.",
inputs: vec![field_req(
"escalation_id",
TypeSchema::String,
"Escalation ID.",
)],
outputs: vec![field("result", TypeSchema::Json, "Dismissal confirmation.")],
},
_other => ControllerSchema {
namespace: "subconscious",
function: "unknown",
description: "Unknown subconscious controller function.",
inputs: vec![FieldSchema {
name: "function",
ty: TypeSchema::String,
comment: "Unknown function requested.",
required: true,
}],
outputs: vec![FieldSchema {
name: "error",
ty: TypeSchema::String,
comment: "Error details.",
required: true,
}],
description: "Unknown subconscious function.",
inputs: vec![],
outputs: vec![field("error", TypeSchema::String, "Error details.")],
},
}
}
// ── Handlers ─────────────────────────────────────────────────────────────────
fn handle_status(_params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let lock = get_or_init_engine().await?;
let guard = lock.lock().await;
let engine = guard
.as_ref()
.ok_or_else(|| "engine not initialized".to_string())?;
let status = engine.status().await;
// Read status entirely from DB — never touch the engine mutex.
// The engine lock is held for the full tick duration, so any RPC
// that acquires it would block until the tick completes.
let config = load_config().await?;
let hb = &config.heartbeat;
let (task_count, pending_escalations, last_tick_at, total_ticks) =
store::with_connection(&config.workspace_dir, |conn| {
let tc = store::task_count(conn).unwrap_or(0);
let pe = store::pending_escalation_count(conn).unwrap_or(0);
let (lt, tt) = conn
.query_row(
"SELECT MAX(tick_at), COUNT(DISTINCT tick_at) FROM subconscious_log",
[],
|row| Ok((row.get::<_, Option<f64>>(0)?, row.get::<_, u64>(1)?)),
)
.unwrap_or((None, 0));
Ok((tc, pe, lt, tt))
})
.map_err(|e| e.to_string())?;
let status = super::types::SubconsciousStatus {
enabled: hb.enabled && hb.inference_enabled,
interval_minutes: hb.interval_minutes.max(5),
last_tick_at,
total_ticks,
task_count,
pending_escalations,
consecutive_failures: 0, // Only available from in-memory state; 0 is fine for UI
};
to_json(RpcOutcome::single_log(status, "subconscious status"))
})
}
@@ -104,72 +240,236 @@ fn handle_status(_params: Map<String, Value>) -> ControllerFuture {
fn handle_trigger(_params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let lock = get_or_init_engine().await?;
let guard = lock.lock().await;
let engine = guard
.as_ref()
.ok_or_else(|| "engine not initialized".to_string())?;
let result = engine.tick().await.map_err(|e| e.to_string())?;
// Spawn the tick in the background so the RPC returns immediately.
// The frontend can poll status/log to see in_progress → final transitions.
let lock_clone = std::sync::Arc::clone(&lock);
tokio::spawn(async move {
let guard = lock_clone.lock().await;
if let Some(engine) = guard.as_ref() {
match engine.tick().await {
Ok(result) => {
tracing::info!(
"[subconscious] manual tick: executed={} escalated={} duration={}ms",
result.executed,
result.escalated,
result.duration_ms
);
}
Err(e) => {
tracing::warn!("[subconscious] manual tick error: {e}");
}
}
}
});
to_json(RpcOutcome::single_log(
result,
"subconscious tick completed",
serde_json::json!({"triggered": true}),
"subconscious tick triggered",
))
})
}
fn handle_actions(params: Map<String, Value>) -> ControllerFuture {
fn handle_tasks_list(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let limit = params.get("limit").and_then(|v| v.as_u64()).unwrap_or(20) as usize;
let enabled_only = params
.get("enabled_only")
.and_then(|v| v.as_bool())
.unwrap_or(false);
let config = load_config().await?;
let tasks = store::with_connection(&config.workspace_dir, |conn| {
store::list_tasks(conn, enabled_only)
})
.map_err(|e| e.to_string())?;
to_json(RpcOutcome::single_log(tasks, "tasks listed"))
})
}
let config = crate::openhuman::config::Config::load_or_init()
fn handle_tasks_add(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let title = params
.get("title")
.and_then(|v| v.as_str())
.ok_or("title is required")?
.to_string();
let source = match params.get("source").and_then(|v| v.as_str()) {
Some("system") => TaskSource::System,
_ => TaskSource::User,
};
let lock = get_or_init_engine().await?;
let guard = lock.lock().await;
let engine = guard.as_ref().ok_or("engine not initialized")?;
let task = engine
.add_task(&title, source)
.await
.map_err(|e| format!("load config: {e}"))?;
let memory =
crate::openhuman::memory::MemoryClient::from_workspace_dir(config.workspace_dir)
.map_err(|e| format!("memory client: {e}"))?;
let entries = memory
.kv_list_namespace("subconscious")
.await
.map_err(|e| format!("list actions: {e}"))?;
let mut actions: Vec<Value> = Vec::new();
for entry in entries {
let key = entry.get("key").and_then(|v| v.as_str()).unwrap_or("");
if !key.starts_with("actions:") {
continue;
}
// Key format: "actions:{ms}:{suffix}" — extract timestamp before suffix
let timestamp = key
.strip_prefix("actions:")
.and_then(|s| s.split(':').next())
.and_then(|s| s.parse::<f64>().ok())
.unwrap_or(0.0);
let value = entry.get("value").and_then(|v| v.as_str()).unwrap_or("[]");
let parsed_actions: Value =
serde_json::from_str(value).unwrap_or(Value::String(value.to_string()));
actions.push(serde_json::json!({
"tick_at": timestamp,
"actions": parsed_actions,
}));
}
actions.sort_by(|a, b| {
let ta = a.get("tick_at").and_then(|v| v.as_f64()).unwrap_or(0.0);
let tb = b.get("tick_at").and_then(|v| v.as_f64()).unwrap_or(0.0);
tb.partial_cmp(&ta).unwrap_or(std::cmp::Ordering::Equal)
});
actions.truncate(limit);
.map_err(|e| e.to_string())?;
to_json(RpcOutcome::single_log(task, "task added"))
})
}
fn handle_tasks_update(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let task_id = params
.get("task_id")
.and_then(|v| v.as_str())
.ok_or("task_id is required")?
.to_string();
let patch = TaskPatch {
title: params
.get("title")
.and_then(|v| v.as_str())
.map(String::from),
recurrence: params.get("recurrence").and_then(|v| v.as_str()).map(|s| {
if s == "once" {
TaskRecurrence::Once
} else if let Some(expr) = s.strip_prefix("cron:") {
TaskRecurrence::Cron(expr.to_string())
} else {
TaskRecurrence::Pending
}
}),
enabled: params.get("enabled").and_then(|v| v.as_bool()),
};
let config = load_config().await?;
store::with_connection(&config.workspace_dir, |conn| {
store::update_task(conn, &task_id, &patch)
})
.map_err(|e| e.to_string())?;
to_json(RpcOutcome::single_log(
serde_json::json!({ "entries": actions, "count": actions.len() }),
"subconscious actions listed",
serde_json::json!({"updated": task_id}),
"task updated",
))
})
}
fn handle_tasks_remove(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let task_id = params
.get("task_id")
.and_then(|v| v.as_str())
.ok_or("task_id is required")?
.to_string();
let config = load_config().await?;
store::with_connection(&config.workspace_dir, |conn| {
store::remove_task(conn, &task_id)
})
.map_err(|e| e.to_string())?;
to_json(RpcOutcome::single_log(
serde_json::json!({"removed": task_id}),
"task removed",
))
})
}
fn handle_log_list(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let task_id = params.get("task_id").and_then(|v| v.as_str());
let limit = params.get("limit").and_then(|v| v.as_u64()).unwrap_or(50) as usize;
let config = load_config().await?;
let entries = store::with_connection(&config.workspace_dir, |conn| {
store::list_log_entries(conn, task_id, limit)
})
.map_err(|e| e.to_string())?;
to_json(RpcOutcome::single_log(entries, "log entries listed"))
})
}
fn handle_escalations_list(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let status_filter = params
.get("status")
.and_then(|v| v.as_str())
.map(|s| match s {
"approved" => EscalationStatus::Approved,
"dismissed" => EscalationStatus::Dismissed,
_ => EscalationStatus::Pending,
});
let config = load_config().await?;
let escalations = store::with_connection(&config.workspace_dir, |conn| {
store::list_escalations(conn, status_filter.as_ref())
})
.map_err(|e| e.to_string())?;
to_json(RpcOutcome::single_log(escalations, "escalations listed"))
})
}
fn handle_escalations_approve(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let escalation_id = params
.get("escalation_id")
.and_then(|v| v.as_str())
.ok_or("escalation_id is required")?
.to_string();
let lock = get_or_init_engine().await?;
let guard = lock.lock().await;
let engine = guard.as_ref().ok_or("engine not initialized")?;
engine
.approve_escalation(&escalation_id)
.await
.map_err(|e| e.to_string())?;
to_json(RpcOutcome::single_log(
serde_json::json!({"approved": escalation_id}),
"escalation approved and executed",
))
})
}
fn handle_escalations_dismiss(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let escalation_id = params
.get("escalation_id")
.and_then(|v| v.as_str())
.ok_or("escalation_id is required")?
.to_string();
let lock = get_or_init_engine().await?;
let guard = lock.lock().await;
let engine = guard.as_ref().ok_or("engine not initialized")?;
engine
.dismiss_escalation(&escalation_id)
.await
.map_err(|e| e.to_string())?;
to_json(RpcOutcome::single_log(
serde_json::json!({"dismissed": escalation_id}),
"escalation dismissed",
))
})
}
// ── Helpers ──────────────────────────────────────────────────────────────────
async fn load_config() -> Result<crate::openhuman::config::Config, String> {
crate::openhuman::config::Config::load_or_init()
.await
.map_err(|e| format!("load config: {e}"))
}
fn field(name: &'static str, ty: TypeSchema, comment: &'static str) -> FieldSchema {
FieldSchema {
name,
ty,
comment,
required: true,
}
}
fn field_req(name: &'static str, ty: TypeSchema, comment: &'static str) -> FieldSchema {
FieldSchema {
name,
ty,
comment,
required: true,
}
}
fn field_opt(name: &'static str, ty: TypeSchema, comment: &'static str) -> FieldSchema {
FieldSchema {
name,
ty,
comment,
required: false,
}
}
fn to_json<T: serde::Serialize>(outcome: RpcOutcome<T>) -> Result<Value, String> {
outcome.into_cli_compatible_json()
}
+10 -14
View File
@@ -1,6 +1,6 @@
//! Assembles a bounded "situation report" for the subconscious tick.
//! Gathers deltas since the last tick from memory, tools, environment,
//! and HEARTBEAT.md — capped at the configured token budget.
//! and pending tasks — capped at the configured token budget.
use crate::openhuman::memory::MemoryClient;
use std::fmt::Write;
@@ -27,8 +27,8 @@ pub async fn build_situation_report(
let env_section = build_environment_section(workspace_dir);
append_section(&mut report, &mut remaining, &env_section);
// Section 2: HEARTBEAT.md tasks
let tasks_section = build_tasks_section(workspace_dir).await;
// Section 2: Pending tasks from SQLite
let tasks_section = build_tasks_section(workspace_dir);
append_section(&mut report, &mut remaining, &tasks_section);
// Section 3: Memory documents (delta since last tick)
@@ -74,25 +74,21 @@ fn build_environment_section(workspace_dir: &Path) -> String {
)
}
async fn build_tasks_section(workspace_dir: &Path) -> String {
let heartbeat_path = workspace_dir.join("HEARTBEAT.md");
let content = match tokio::fs::read_to_string(&heartbeat_path).await {
Ok(c) => c,
Err(_) => return "## Pending Tasks\n\nNo HEARTBEAT.md found.\n".to_string(),
fn build_tasks_section(workspace_dir: &Path) -> String {
let tasks = match super::store::with_connection(workspace_dir, |conn| {
super::store::list_tasks(conn, false)
}) {
Ok(tasks) => tasks,
Err(_) => return "## Pending Tasks\n\nFailed to read tasks.\n".to_string(),
};
let tasks: Vec<&str> = content
.lines()
.filter_map(|line| line.trim().strip_prefix("- "))
.collect();
if tasks.is_empty() {
return "## Pending Tasks\n\nNo tasks defined.\n".to_string();
}
let mut section = String::from("## Pending Tasks\n\n");
for task in &tasks {
let _ = writeln!(section, "- {task}");
let _ = writeln!(section, "- {}", task.title);
}
section
}
+716
View File
@@ -0,0 +1,716 @@
//! SQLite persistence for subconscious tasks, execution log, and escalations.
//!
//! Follows the cron module's `with_connection` pattern: opens the database,
//! runs DDL on every connection, and provides pure functions.
use anyhow::{Context, Result};
use rusqlite::Connection;
use std::path::Path;
use uuid::Uuid;
use super::types::{
Escalation, EscalationPriority, EscalationStatus, SubconsciousLogEntry, SubconsciousTask,
TaskPatch, TaskRecurrence, TaskSource,
};
/// Open the subconscious database and run schema migrations.
pub fn with_connection<T>(
workspace_dir: &Path,
f: impl FnOnce(&Connection) -> Result<T>,
) -> Result<T> {
let db_path = workspace_dir.join("subconscious").join("subconscious.db");
if let Some(parent) = db_path.parent() {
std::fs::create_dir_all(parent)
.with_context(|| format!("failed to create subconscious dir: {}", parent.display()))?;
}
let conn = Connection::open(&db_path)
.with_context(|| format!("failed to open subconscious DB: {}", db_path.display()))?;
conn.execute_batch(SCHEMA_DDL)
.with_context(|| "failed to run subconscious schema DDL")?;
f(&conn)
}
const SCHEMA_DDL: &str = "
PRAGMA foreign_keys = ON;
PRAGMA journal_mode = WAL;
CREATE TABLE IF NOT EXISTS subconscious_tasks (
id TEXT PRIMARY KEY,
title TEXT NOT NULL,
source TEXT NOT NULL DEFAULT 'user',
recurrence TEXT NOT NULL DEFAULT 'pending',
enabled INTEGER NOT NULL DEFAULT 1,
last_run_at REAL,
next_run_at REAL,
completed INTEGER NOT NULL DEFAULT 0,
created_at REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_tasks_next_run
ON subconscious_tasks(next_run_at);
CREATE INDEX IF NOT EXISTS idx_tasks_enabled
ON subconscious_tasks(enabled, completed);
CREATE TABLE IF NOT EXISTS subconscious_log (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL,
tick_at REAL NOT NULL,
decision TEXT NOT NULL,
result TEXT,
duration_ms INTEGER,
created_at REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_log_task
ON subconscious_log(task_id, tick_at DESC);
CREATE INDEX IF NOT EXISTS idx_log_tick
ON subconscious_log(tick_at DESC);
CREATE TABLE IF NOT EXISTS subconscious_escalations (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL,
log_id TEXT,
title TEXT NOT NULL,
description TEXT NOT NULL,
priority TEXT NOT NULL DEFAULT 'normal',
status TEXT NOT NULL DEFAULT 'pending',
created_at REAL NOT NULL,
resolved_at REAL
);
CREATE INDEX IF NOT EXISTS idx_escalations_status
ON subconscious_escalations(status);
";
// ── Task CRUD ────────────────────────────────────────────────────────────────
pub fn add_task(
conn: &Connection,
title: &str,
source: TaskSource,
recurrence: TaskRecurrence,
) -> Result<SubconsciousTask> {
let id = Uuid::new_v4().to_string();
let now = now_secs();
let source_str = serde_json::to_value(&source)
.unwrap_or_default()
.as_str()
.unwrap_or("user")
.to_string();
let recurrence_str = recurrence_to_string(&recurrence);
conn.execute(
"INSERT INTO subconscious_tasks (id, title, source, recurrence, created_at)
VALUES (?1, ?2, ?3, ?4, ?5)",
rusqlite::params![id, title, source_str, recurrence_str, now],
)?;
Ok(SubconsciousTask {
id,
title: title.to_string(),
source,
recurrence,
enabled: true,
last_run_at: None,
next_run_at: None,
completed: false,
created_at: now,
})
}
pub fn get_task(conn: &Connection, task_id: &str) -> Result<SubconsciousTask> {
conn.query_row(
"SELECT id, title, source, recurrence, enabled, last_run_at, next_run_at, completed, created_at
FROM subconscious_tasks WHERE id = ?1",
[task_id],
row_to_task,
)
.with_context(|| format!("task not found: {task_id}"))
}
pub fn list_tasks(conn: &Connection, enabled_only: bool) -> Result<Vec<SubconsciousTask>> {
let sql = if enabled_only {
"SELECT id, title, source, recurrence, enabled, last_run_at, next_run_at, completed, created_at
FROM subconscious_tasks WHERE enabled = 1 ORDER BY created_at"
} else {
"SELECT id, title, source, recurrence, enabled, last_run_at, next_run_at, completed, created_at
FROM subconscious_tasks ORDER BY created_at"
};
let mut stmt = conn.prepare(sql)?;
let tasks = stmt
.query_map([], row_to_task)?
.collect::<Result<Vec<_>, _>>()?;
Ok(tasks)
}
pub fn update_task(conn: &Connection, task_id: &str, patch: &TaskPatch) -> Result<()> {
if let Some(ref title) = patch.title {
conn.execute(
"UPDATE subconscious_tasks SET title = ?1 WHERE id = ?2",
rusqlite::params![title, task_id],
)?;
}
if let Some(ref recurrence) = patch.recurrence {
conn.execute(
"UPDATE subconscious_tasks SET recurrence = ?1 WHERE id = ?2",
rusqlite::params![recurrence_to_string(recurrence), task_id],
)?;
}
if let Some(enabled) = patch.enabled {
conn.execute(
"UPDATE subconscious_tasks SET enabled = ?1 WHERE id = ?2",
rusqlite::params![enabled, task_id],
)?;
}
Ok(())
}
/// Remove a task. System tasks cannot be deleted — only disabled.
pub fn remove_task(conn: &Connection, task_id: &str) -> Result<()> {
let source: String = conn
.query_row(
"SELECT source FROM subconscious_tasks WHERE id = ?1",
[task_id],
|row| row.get(0),
)
.with_context(|| format!("task not found: {task_id}"))?;
if source == "system" {
anyhow::bail!("System tasks cannot be deleted. Disable them instead.");
}
conn.execute("DELETE FROM subconscious_tasks WHERE id = ?1", [task_id])?;
Ok(())
}
/// Get tasks that are due for evaluation (enabled, not completed, due now or never run).
pub fn due_tasks(conn: &Connection, now: f64) -> Result<Vec<SubconsciousTask>> {
let mut stmt = conn.prepare(
"SELECT id, title, source, recurrence, enabled, last_run_at, next_run_at, completed, created_at
FROM subconscious_tasks
WHERE enabled = 1 AND completed = 0
AND (next_run_at IS NULL OR next_run_at <= ?1)
ORDER BY next_run_at NULLS FIRST",
)?;
let tasks = stmt
.query_map([now], row_to_task)?
.collect::<Result<Vec<_>, _>>()?;
Ok(tasks)
}
pub fn mark_task_completed(conn: &Connection, task_id: &str) -> Result<()> {
conn.execute(
"UPDATE subconscious_tasks SET completed = 1 WHERE id = ?1",
[task_id],
)?;
Ok(())
}
pub fn update_task_run_times(
conn: &Connection,
task_id: &str,
last_run_at: f64,
next_run_at: Option<f64>,
) -> Result<()> {
conn.execute(
"UPDATE subconscious_tasks SET last_run_at = ?1, next_run_at = ?2 WHERE id = ?3",
rusqlite::params![last_run_at, next_run_at, task_id],
)?;
Ok(())
}
pub fn task_count(conn: &Connection) -> Result<u64> {
conn.query_row(
"SELECT COUNT(*) FROM subconscious_tasks WHERE enabled = 1 AND completed = 0",
[],
|row| row.get::<_, u64>(0),
)
.map_err(Into::into)
}
// ── Log CRUD ─────────────────────────────────────────────────────────────────
pub fn add_log_entry(
conn: &Connection,
task_id: &str,
tick_at: f64,
decision: &str,
result: Option<&str>,
duration_ms: Option<i64>,
) -> Result<SubconsciousLogEntry> {
let id = Uuid::new_v4().to_string();
let now = now_secs();
conn.execute(
"INSERT INTO subconscious_log (id, task_id, tick_at, decision, result, duration_ms, created_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
rusqlite::params![id, task_id, tick_at, decision, result, duration_ms, now],
)?;
Ok(SubconsciousLogEntry {
id,
task_id: task_id.to_string(),
tick_at,
decision: decision.to_string(),
result: result.map(String::from),
duration_ms,
created_at: now,
})
}
/// Update an existing log entry's decision, result, and duration in place.
pub fn update_log_entry(
conn: &Connection,
log_id: &str,
decision: &str,
result: Option<&str>,
duration_ms: Option<i64>,
) -> Result<()> {
conn.execute(
"UPDATE subconscious_log SET decision = ?1, result = ?2, duration_ms = ?3 WHERE id = ?4",
rusqlite::params![decision, result, duration_ms, log_id],
)?;
Ok(())
}
/// Bulk-update ALL in_progress log entries to cancelled.
/// Any entry still in_progress when a new tick starts is stale by definition.
pub fn cancel_stale_in_progress(conn: &Connection) -> Result<usize> {
let count = conn.execute(
"UPDATE subconscious_log SET decision = 'cancelled', result = 'Superseded by new tick'
WHERE decision = 'in_progress'",
[],
)?;
Ok(count)
}
pub fn list_log_entries(
conn: &Connection,
task_id: Option<&str>,
limit: usize,
) -> Result<Vec<SubconsciousLogEntry>> {
let (sql, params): (&str, Vec<Box<dyn rusqlite::types::ToSql>>) = if let Some(tid) = task_id {
(
"SELECT id, task_id, tick_at, decision, result, duration_ms, created_at
FROM subconscious_log WHERE task_id = ?1 ORDER BY tick_at DESC LIMIT ?2",
vec![Box::new(tid.to_string()), Box::new(limit as i64)],
)
} else {
(
"SELECT id, task_id, tick_at, decision, result, duration_ms, created_at
FROM subconscious_log ORDER BY tick_at DESC LIMIT ?1",
vec![Box::new(limit as i64)],
)
};
let mut stmt = conn.prepare(sql)?;
let entries = stmt
.query_map(rusqlite::params_from_iter(params.iter()), |row| {
Ok(SubconsciousLogEntry {
id: row.get(0)?,
task_id: row.get(1)?,
tick_at: row.get(2)?,
decision: row.get(3)?,
result: row.get(4)?,
duration_ms: row.get(5)?,
created_at: row.get(6)?,
})
})?
.collect::<Result<Vec<_>, _>>()?;
Ok(entries)
}
// ── Escalation CRUD ──────────────────────────────────────────────────────────
pub fn add_escalation(
conn: &Connection,
task_id: &str,
log_id: Option<&str>,
title: &str,
description: &str,
priority: &EscalationPriority,
) -> Result<Escalation> {
let id = Uuid::new_v4().to_string();
let now = now_secs();
let priority_str = serde_json::to_value(priority)
.unwrap_or_default()
.as_str()
.unwrap_or("normal")
.to_string();
conn.execute(
"INSERT INTO subconscious_escalations (id, task_id, log_id, title, description, priority, created_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
rusqlite::params![id, task_id, log_id, title, description, priority_str, now],
)?;
Ok(Escalation {
id,
task_id: task_id.to_string(),
log_id: log_id.map(String::from),
title: title.to_string(),
description: description.to_string(),
priority: priority.clone(),
status: EscalationStatus::Pending,
created_at: now,
resolved_at: None,
})
}
pub fn list_escalations(
conn: &Connection,
status_filter: Option<&EscalationStatus>,
) -> Result<Vec<Escalation>> {
let (sql, params): (&str, Vec<Box<dyn rusqlite::types::ToSql>>) = if let Some(status) =
status_filter
{
let status_str = serde_json::to_value(status)
.unwrap_or_default()
.as_str()
.unwrap_or("pending")
.to_string();
(
"SELECT id, task_id, log_id, title, description, priority, status, created_at, resolved_at
FROM subconscious_escalations WHERE status = ?1 ORDER BY created_at DESC",
vec![Box::new(status_str)],
)
} else {
(
"SELECT id, task_id, log_id, title, description, priority, status, created_at, resolved_at
FROM subconscious_escalations ORDER BY created_at DESC",
vec![],
)
};
let mut stmt = conn.prepare(sql)?;
let rows = stmt
.query_map(rusqlite::params_from_iter(params.iter()), row_to_escalation)?
.collect::<Result<Vec<_>, _>>()?;
Ok(rows)
}
pub fn resolve_escalation(
conn: &Connection,
escalation_id: &str,
status: &EscalationStatus,
) -> Result<()> {
let now = now_secs();
let status_str = serde_json::to_value(status)
.unwrap_or_default()
.as_str()
.unwrap_or("dismissed")
.to_string();
conn.execute(
"UPDATE subconscious_escalations SET status = ?1, resolved_at = ?2 WHERE id = ?3",
rusqlite::params![status_str, now, escalation_id],
)?;
Ok(())
}
pub fn pending_escalation_count(conn: &Connection) -> Result<u64> {
conn.query_row(
"SELECT COUNT(*) FROM subconscious_escalations WHERE status = 'pending'",
[],
|row| row.get::<_, u64>(0),
)
.map_err(Into::into)
}
pub fn get_escalation(conn: &Connection, escalation_id: &str) -> Result<Escalation> {
conn.query_row(
"SELECT id, task_id, log_id, title, description, priority, status, created_at, resolved_at
FROM subconscious_escalations WHERE id = ?1",
[escalation_id],
row_to_escalation,
)
.with_context(|| format!("escalation not found: {escalation_id}"))
}
// ── Seed default system tasks ────────────────────────────────────────────────
/// Default system tasks that are always seeded and cannot be deleted.
const DEFAULT_SYSTEM_TASKS: &[&str] = &[
"Check connected skills for errors or disconnections",
"Review new memory updates for actionable items",
"Monitor system health (Ollama, memory, connections)",
];
/// Seed default system tasks into SQLite.
/// Skips tasks whose title already exists. Returns the count of newly created tasks.
pub fn seed_default_tasks(conn: &Connection) -> Result<usize> {
let mut count = 0;
for title in DEFAULT_SYSTEM_TASKS {
if !task_title_exists(conn, title)? {
add_task(conn, title, TaskSource::System, TaskRecurrence::Pending)?;
count += 1;
}
}
Ok(count)
}
fn task_title_exists(conn: &Connection, title: &str) -> Result<bool> {
Ok(conn.query_row(
"SELECT EXISTS(SELECT 1 FROM subconscious_tasks WHERE title = ?1)",
[title],
|row| row.get(0),
)?)
}
// ── Helpers ──────────────────────────────────────────────────────────────────
fn row_to_task(row: &rusqlite::Row) -> rusqlite::Result<SubconsciousTask> {
let source_str: String = row.get(2)?;
let recurrence_str: String = row.get(3)?;
Ok(SubconsciousTask {
id: row.get(0)?,
title: row.get(1)?,
source: string_to_source(&source_str),
recurrence: string_to_recurrence(&recurrence_str),
enabled: row.get::<_, bool>(4)?,
last_run_at: row.get(5)?,
next_run_at: row.get(6)?,
completed: row.get::<_, bool>(7)?,
created_at: row.get(8)?,
})
}
fn row_to_escalation(row: &rusqlite::Row) -> rusqlite::Result<Escalation> {
let priority_str: String = row.get(5)?;
let status_str: String = row.get(6)?;
Ok(Escalation {
id: row.get(0)?,
task_id: row.get(1)?,
log_id: row.get(2)?,
title: row.get(3)?,
description: row.get(4)?,
priority: string_to_priority(&priority_str),
status: string_to_status(&status_str),
created_at: row.get(7)?,
resolved_at: row.get(8)?,
})
}
fn recurrence_to_string(r: &TaskRecurrence) -> String {
match r {
TaskRecurrence::Once => "once".to_string(),
TaskRecurrence::Cron(expr) => format!("cron:{expr}"),
TaskRecurrence::Pending => "pending".to_string(),
}
}
fn string_to_recurrence(s: &str) -> TaskRecurrence {
if s == "once" {
TaskRecurrence::Once
} else if let Some(expr) = s.strip_prefix("cron:") {
TaskRecurrence::Cron(expr.to_string())
} else {
TaskRecurrence::Pending
}
}
fn string_to_source(s: &str) -> TaskSource {
match s {
"system" => TaskSource::System,
_ => TaskSource::User,
}
}
fn string_to_priority(s: &str) -> EscalationPriority {
match s {
"critical" => EscalationPriority::Critical,
"important" => EscalationPriority::Important,
_ => EscalationPriority::Normal,
}
}
fn string_to_status(s: &str) -> EscalationStatus {
match s {
"approved" => EscalationStatus::Approved,
"dismissed" => EscalationStatus::Dismissed,
_ => EscalationStatus::Pending,
}
}
fn now_secs() -> f64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs_f64())
.unwrap_or(0.0)
}
/// Compute the next run time for a cron expression.
/// Normalizes 5-field cron to 6-field (prepends seconds=0) for the `cron` crate.
pub fn compute_next_run(cron_expr: &str) -> Option<f64> {
let normalized = normalize_cron_expr(cron_expr);
let schedule = normalized.parse::<cron::Schedule>().ok()?;
let next = schedule.upcoming(chrono::Utc).next()?;
Some(next.timestamp() as f64)
}
fn normalize_cron_expr(expr: &str) -> String {
let fields: Vec<&str> = expr.split_whitespace().collect();
if fields.len() == 5 {
format!("0 {expr}")
} else {
expr.to_string()
}
}
#[cfg(test)]
mod tests {
use super::*;
fn test_conn() -> Connection {
let conn = Connection::open_in_memory().unwrap();
conn.execute_batch(SCHEMA_DDL).unwrap();
conn
}
#[test]
fn crud_tasks() {
let conn = test_conn();
let task = add_task(&conn, "Check email", TaskSource::User, TaskRecurrence::Once).unwrap();
assert_eq!(task.title, "Check email");
assert!(!task.completed);
let fetched = get_task(&conn, &task.id).unwrap();
assert_eq!(fetched.title, "Check email");
let all = list_tasks(&conn, false).unwrap();
assert_eq!(all.len(), 1);
update_task(
&conn,
&task.id,
&TaskPatch {
title: Some("Check Gmail".into()),
..Default::default()
},
)
.unwrap();
let updated = get_task(&conn, &task.id).unwrap();
assert_eq!(updated.title, "Check Gmail");
mark_task_completed(&conn, &task.id).unwrap();
let done = get_task(&conn, &task.id).unwrap();
assert!(done.completed);
remove_task(&conn, &task.id).unwrap();
assert!(get_task(&conn, &task.id).is_err());
}
#[test]
fn due_tasks_filters_correctly() {
let conn = test_conn();
let now = now_secs();
// Task with no next_run_at — should be due
add_task(
&conn,
"No schedule",
TaskSource::User,
TaskRecurrence::Pending,
)
.unwrap();
// Task with future next_run_at — should NOT be due
let future_task =
add_task(&conn, "Future task", TaskSource::User, TaskRecurrence::Once).unwrap();
update_task_run_times(&conn, &future_task.id, now, Some(now + 3600.0)).unwrap();
// Task with past next_run_at — should be due
let past_task =
add_task(&conn, "Past due", TaskSource::User, TaskRecurrence::Once).unwrap();
update_task_run_times(&conn, &past_task.id, now - 7200.0, Some(now - 3600.0)).unwrap();
let due = due_tasks(&conn, now).unwrap();
assert_eq!(due.len(), 2); // "No schedule" + "Past due"
assert!(due.iter().any(|t| t.title == "No schedule"));
assert!(due.iter().any(|t| t.title == "Past due"));
assert!(!due.iter().any(|t| t.title == "Future task"));
}
#[test]
fn crud_log_entries() {
let conn = test_conn();
let task = add_task(&conn, "Test", TaskSource::User, TaskRecurrence::Once).unwrap();
let now = now_secs();
let entry = add_log_entry(
&conn,
&task.id,
now,
"act",
Some("Did the thing"),
Some(150),
)
.unwrap();
assert_eq!(entry.decision, "act");
let entries = list_log_entries(&conn, Some(&task.id), 10).unwrap();
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].result.as_deref(), Some("Did the thing"));
let all_entries = list_log_entries(&conn, None, 10).unwrap();
assert_eq!(all_entries.len(), 1);
}
#[test]
fn crud_escalations() {
let conn = test_conn();
let task = add_task(&conn, "Test", TaskSource::User, TaskRecurrence::Once).unwrap();
let esc = add_escalation(
&conn,
&task.id,
None,
"Deadline moved",
"The API deadline was moved to tomorrow",
&EscalationPriority::Critical,
)
.unwrap();
assert_eq!(esc.status, EscalationStatus::Pending);
let pending = list_escalations(&conn, Some(&EscalationStatus::Pending)).unwrap();
assert_eq!(pending.len(), 1);
assert_eq!(pending_escalation_count(&conn).unwrap(), 1);
resolve_escalation(&conn, &esc.id, &EscalationStatus::Approved).unwrap();
let resolved = get_escalation(&conn, &esc.id).unwrap();
assert_eq!(resolved.status, EscalationStatus::Approved);
assert!(resolved.resolved_at.is_some());
assert_eq!(pending_escalation_count(&conn).unwrap(), 0);
}
#[test]
fn seed_default_tasks_creates_system_tasks() {
let conn = test_conn();
let count = seed_default_tasks(&conn).unwrap();
assert_eq!(count, DEFAULT_SYSTEM_TASKS.len());
// Second seed should not duplicate
let count2 = seed_default_tasks(&conn).unwrap();
assert_eq!(count2, 0);
let tasks = list_tasks(&conn, false).unwrap();
assert_eq!(tasks.len(), DEFAULT_SYSTEM_TASKS.len());
assert!(tasks.iter().all(|t| t.source == TaskSource::System));
}
#[test]
fn recurrence_roundtrip() {
assert_eq!(
string_to_recurrence(&recurrence_to_string(&TaskRecurrence::Once)),
TaskRecurrence::Once
);
assert_eq!(
string_to_recurrence(&recurrence_to_string(&TaskRecurrence::Pending)),
TaskRecurrence::Pending
);
assert_eq!(
string_to_recurrence(&recurrence_to_string(&TaskRecurrence::Cron(
"0 8 * * *".into()
))),
TaskRecurrence::Cron("0 8 * * *".into())
);
}
}
+136 -48
View File
@@ -1,71 +1,159 @@
//! Type definitions for the subconscious task execution system.
use serde::{Deserialize, Serialize};
/// Decision produced by the subconscious tick.
// ── Task types ───────────────────────────────────────────────────────────────
/// A task managed by the subconscious engine.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SubconsciousTask {
pub id: String,
pub title: String,
pub source: TaskSource,
pub recurrence: TaskRecurrence,
pub enabled: bool,
pub last_run_at: Option<f64>,
pub next_run_at: Option<f64>,
pub completed: bool,
pub created_at: f64,
}
/// Where the task came from.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Decision {
/// Nothing meaningful changed — skip.
pub enum TaskSource {
/// Auto-populated by the system (skills health, Ollama status, etc.)
System,
/// Added by the user via UI or agent.
User,
}
/// How often the task should run.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TaskRecurrence {
/// Execute once, then mark completed.
Once,
/// Recurrent on a cron schedule (5-field expression).
Cron(String),
/// Not yet classified — agent will decide on first tick.
Pending,
}
/// Partial update for a task.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct TaskPatch {
pub title: Option<String>,
pub recurrence: Option<TaskRecurrence>,
pub enabled: Option<bool>,
}
// ── Tick evaluation types ────────────────────────────────────────────────────
/// Per-tick decision for a single task.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TickDecision {
/// Nothing relevant in current state for this task.
#[default]
Noop,
/// Something changed that can be handled locally (store memory, update state).
/// State has something relevant — execute the task.
Act,
/// Something important changed that requires the full agent.
/// Ambiguous or risky — surface to user for approval.
Escalate,
}
/// A single action recommended by the subconscious.
/// The local model's evaluation of a single task against the current state.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RecommendedAction {
#[serde(rename = "type")]
pub action_type: ActionType,
pub description: String,
pub priority: Priority,
/// Which HEARTBEAT.md task this action relates to.
#[serde(default)]
pub task: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ActionType {
Notify,
StoreMemory,
EscalateToAgent,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Priority {
Low,
Medium,
High,
}
/// Output from the local model after evaluating the situation report.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TickOutput {
pub decision: Decision,
pub struct TaskEvaluation {
pub task_id: String,
pub decision: TickDecision,
pub reason: String,
#[serde(default)]
pub actions: Vec<RecommendedAction>,
}
/// Result of a single subconscious tick, including metadata.
/// Full evaluation response from the local model.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TickResult {
pub tick_at: f64,
pub output: TickOutput,
pub source_doc_ids: Vec<String>,
pub duration_ms: u64,
pub tokens_used: u64,
pub struct EvaluationResponse {
pub evaluations: Vec<TaskEvaluation>,
}
/// Summary of the subconscious loop status.
// ── Execution types ──────────────────────────────────────────────────────────
/// Result of executing a single task.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutionResult {
pub output: String,
pub used_tools: bool,
pub duration_ms: u64,
}
// ── Log types ────────────────────────────────────────────────────────────────
/// A single entry in the execution log.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SubconsciousLogEntry {
pub id: String,
pub task_id: String,
pub tick_at: f64,
pub decision: String,
pub result: Option<String>,
pub duration_ms: Option<i64>,
pub created_at: f64,
}
// ── Escalation types ─────────────────────────────────────────────────────────
/// An escalation waiting for user input.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Escalation {
pub id: String,
pub task_id: String,
pub log_id: Option<String>,
pub title: String,
pub description: String,
pub priority: EscalationPriority,
pub status: EscalationStatus,
pub created_at: f64,
pub resolved_at: Option<f64>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EscalationPriority {
Critical,
Important,
Normal,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EscalationStatus {
Pending,
Approved,
Dismissed,
}
// ── Status types ─────────────────────────────────────────────────────────────
/// Summary of the subconscious engine status.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SubconsciousStatus {
pub enabled: bool,
pub interval_minutes: u32,
pub last_tick_at: Option<f64>,
pub last_decision: Option<Decision>,
pub total_ticks: u64,
pub total_escalations: u64,
pub task_count: u64,
pub pending_escalations: u64,
/// Number of consecutive tick failures (resets on success).
pub consecutive_failures: u64,
}
/// Result of a single subconscious tick.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TickResult {
pub tick_at: f64,
pub evaluations: Vec<TaskEvaluation>,
pub executed: usize,
pub escalated: usize,
pub duration_ms: u64,
}
+290
View File
@@ -0,0 +1,290 @@
//! End-to-end subconscious test with real Ollama, real memory, real SQLite.
//!
//! Requires Ollama running at localhost:11434 with a model loaded.
//! Run with: `cargo test --test subconscious_e2e -- --nocapture --ignored`
use std::sync::Arc;
use serde_json::json;
/// Config that skips GLiNER relex model (avoids ORT init).
fn ci_safe_ingestion_config() -> openhuman_core::openhuman::memory::MemoryIngestionConfig {
openhuman_core::openhuman::memory::MemoryIngestionConfig {
model_name: "__test_no_model__".to_string(),
..Default::default()
}
}
async fn ingest_doc(
memory: &openhuman_core::openhuman::memory::UnifiedMemory,
namespace: &str,
key: &str,
title: &str,
content: &str,
) -> String {
use openhuman_core::openhuman::memory::{MemoryIngestionRequest, NamespaceDocumentInput};
let result = memory
.ingest_document(MemoryIngestionRequest {
document: NamespaceDocumentInput {
namespace: namespace.to_string(),
key: key.to_string(),
title: title.to_string(),
content: content.to_string(),
source_type: "test".to_string(),
priority: "high".to_string(),
tags: Vec::new(),
metadata: json!({}),
category: "core".to_string(),
session_id: None,
document_id: None,
},
config: ci_safe_ingestion_config(),
})
.await
.expect("ingest should succeed");
result.document_id
}
/// Full two-tick E2E test:
///
/// **Tick 1**: Gmail has 3 urgent emails, Notion has a deadline tracker.
/// → Ollama should detect urgent items → act or escalate.
///
/// **Tick 2**: New data — deadline moved, ownership changed.
/// → Ollama should detect the change → act or escalate on new state.
///
/// Verifies:
/// - Tasks loaded from HEARTBEAT.md seed
/// - Real Ollama evaluation produces valid decisions
/// - SQLite log entries created for each tick
/// - Act tasks produce text output from Ollama
/// - Second tick sees delta (new data only)
#[tokio::test]
#[ignore] // requires running Ollama
async fn two_tick_e2e_with_real_ollama() {
use openhuman_core::openhuman::memory::embeddings::NoopEmbedding;
use openhuman_core::openhuman::memory::{MemoryClient, UnifiedMemory};
use openhuman_core::openhuman::subconscious::store;
// ── Setup workspace ──────────────────────────────────────────────
let tmp = tempfile::tempdir().expect("tempdir");
let workspace = tmp.path();
// Write HEARTBEAT.md
std::fs::write(
workspace.join("HEARTBEAT.md"),
"\
# Periodic Tasks
- Check Gmail for urgent emails that need immediate attention
- Review Notion project tracker for deadline changes
- Monitor connected skills for errors or disconnections
",
)
.expect("write heartbeat");
// Initialize memory
let memory = UnifiedMemory::new(workspace, Arc::new(NoopEmbedding), None).expect("init memory");
let memory_client =
MemoryClient::from_workspace_dir(workspace.to_path_buf()).expect("memory client");
// ── Tick 1: Ingest initial data ──────────────────────────────────
println!("\n============================================================");
println!(" TICK 1: Initial state — urgent emails + project tracker");
println!("============================================================\n");
ingest_doc(
&memory,
"skill-gmail",
"urgent-emails-batch1",
"3 urgent emails in inbox",
"\
Email 1: From alice@partner.com — Subject: URGENT: API contract deadline
Body: The API integration deadline has been moved from Friday to tomorrow (Thursday).
Please confirm you can deliver by end of day. This is blocking the partner launch.
Email 2: From boss@company.com — Subject: Re: Q1 Budget Review
Body: Need your updated numbers by 3pm today. The board meeting is tomorrow morning.
Please prioritize this over other tasks.
Email 3: From security@company.com — Subject: [ACTION REQUIRED] Password expiry
Body: Your corporate password expires in 24 hours. Please update it via the portal
to avoid being locked out of all systems.",
)
.await;
ingest_doc(
&memory,
"skill-notion",
"q1-tracker-v1",
"Q1 Delivery Tracker",
"\
Project: API Integration
Status: In Progress
Deadline: April 5 (Friday)
Owner: You
Dependencies: Partner API docs (received), Internal review (pending)
Notes: On track for Friday delivery. Partner team confirmed their side is ready.
Project: Q1 Budget Report
Status: Draft
Deadline: Today 3pm
Owner: You
Notes: Numbers need updating. Finance sent corrections yesterday.",
)
.await;
// Build engine with real config
let mut config = openhuman_core::openhuman::config::Config::default();
config.workspace_dir = workspace.to_path_buf();
config.heartbeat.enabled = true;
config.heartbeat.inference_enabled = true;
config.heartbeat.interval_minutes = 5;
config.heartbeat.context_budget_tokens = 40_000;
config.local_ai.enabled = true;
let engine = openhuman_core::openhuman::subconscious::SubconsciousEngine::new(
&config,
Some(Arc::new(memory_client)),
);
// Run tick 1
let result1 = engine.tick().await.expect("tick 1 should succeed");
println!("\n--- Tick 1 Results ---");
println!(" Duration: {}ms", result1.duration_ms);
println!(" Evaluations: {}", result1.evaluations.len());
println!(" Executed: {}", result1.executed);
println!(" Escalated: {}", result1.escalated);
for eval in &result1.evaluations {
println!(" [{}] {:?}{}", eval.task_id, eval.decision, eval.reason);
}
// Verify tick 1
assert!(
!result1.evaluations.is_empty(),
"Ollama should produce evaluations for seeded tasks"
);
// Check SQLite log
let log1 = store::with_connection(workspace, |conn| store::list_log_entries(conn, None, 50))
.expect("list log");
println!("\n Log entries after tick 1: {}", log1.len());
for entry in &log1 {
println!(
" [{}] {}{}",
entry.task_id,
entry.decision,
entry.result.as_deref().unwrap_or("(none)")
);
}
assert!(!log1.is_empty(), "Should have log entries after tick 1");
// Check tasks were seeded
let tasks = store::with_connection(workspace, |conn| store::list_tasks(conn, false))
.expect("list tasks");
println!("\n Tasks: {}", tasks.len());
for t in &tasks {
println!(
" [{}] {} (source={:?}, completed={})",
t.id, t.title, t.source, t.completed
);
}
assert_eq!(tasks.len(), 3, "Should have 3 tasks from HEARTBEAT.md");
// ── Tick 2: Ingest NEW data (state change) ──────────────────────
println!("\n============================================================");
println!(" TICK 2: State change — deadline moved, new urgent email");
println!("============================================================\n");
ingest_doc(
&memory,
"skill-gmail",
"urgent-deadline-moved",
"CRITICAL: API deadline moved to TOMORROW",
"\
Email from alice@partner.com — Subject: RE: URGENT: API contract deadline
Body: Update — the deadline has been moved UP to tomorrow (Wednesday) not Thursday.
The partner CEO is flying in Wednesday evening and wants a demo.
This is now the #1 priority. Please drop everything else.
Email from boss@company.com — Subject: Re: Re: Q1 Budget Review
Body: Good news — finance approved a 1-day extension on the budget report.
New deadline is Friday. Focus on the API delivery instead.",
)
.await;
ingest_doc(
&memory,
"skill-notion",
"q1-tracker-v2",
"Q1 Delivery Tracker (UPDATED)",
"\
Project: API Integration
Status: AT RISK
Deadline: TOMORROW (Wednesday) — moved up from Friday
Owner: You
Dependencies: Partner API docs (received), Internal review (BLOCKING — not started)
Notes: CRITICAL — deadline moved up 2 days. Internal review not started.
Partner CEO demo Wednesday evening. Need to start review NOW.
Blockers: Internal review team not yet assigned.
Project: Q1 Budget Report
Status: Draft
Deadline: Friday (extended from today)
Owner: You
Notes: Extension granted. Lower priority now.",
)
.await;
// Run tick 2
let result2 = engine.tick().await.expect("tick 2 should succeed");
println!("\n--- Tick 2 Results ---");
println!(" Duration: {}ms", result2.duration_ms);
println!(" Evaluations: {}", result2.evaluations.len());
println!(" Executed: {}", result2.executed);
println!(" Escalated: {}", result2.escalated);
for eval in &result2.evaluations {
println!(" [{}] {:?}{}", eval.task_id, eval.decision, eval.reason);
}
// Verify tick 2
assert!(
!result2.evaluations.is_empty(),
"Ollama should produce evaluations for tick 2"
);
// Check cumulative log
let log2 = store::with_connection(workspace, |conn| store::list_log_entries(conn, None, 50))
.expect("list log");
println!("\n Total log entries after tick 2: {}", log2.len());
assert!(
log2.len() > log1.len(),
"Tick 2 should add more log entries"
);
// Check for any escalations
let escalations = store::with_connection(workspace, |conn| store::list_escalations(conn, None))
.expect("list escalations");
println!(" Escalations: {}", escalations.len());
for esc in &escalations {
println!(
" [{}] {}{} (status={:?})",
esc.task_id, esc.title, esc.description, esc.status
);
}
// ── Status check ─────────────────────────────────────────────────
let status = engine.status().await;
println!("\n--- Engine Status ---");
println!(" Enabled: {}", status.enabled);
println!(" Total ticks: {}", status.total_ticks);
println!(" Task count: {}", status.task_count);
println!(" Pending escalations: {}", status.pending_escalations);
assert_eq!(status.total_ticks, 2);
println!("\n============================================================");
println!(" E2E TEST PASSED");
println!("============================================================\n");
}