feat: schedule cache, workflow-aligned prompts, dispatchInbound wakeup

1. ScheduleCache: local cache of today's schedule, synced every 5 min
   from HF backend via new getDaySchedule() API

2. Wakeup prompts updated to reference daily-routine skill workflows
   (task-handson, plan-schedule, slot-complete)

3. Agent wakeup via dispatchInboundMessageWithDispatcher (in-process)
   - Same mechanism as Discord plugin
   - Creates unique session per slot: agent:{agentId}:hf-calendar:slot-{slotId}
   - No WebSocket, CLI, or cron dependency
   - Verified working on test environment

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
operator
2026-04-19 08:10:39 +00:00
parent 248adfaafd
commit ec09578de3
6 changed files with 442 additions and 69 deletions

118
REFACTOR_PLAN.md Normal file
View File

@@ -0,0 +1,118 @@
# CalendarScheduler Refactor Plan (v2)
> Updated 2026-04-19 based on architecture discussion with hang
## Current Issues
1. `process.env.AGENT_ID` doesn't exist in plugins subprocess — always 'unknown'
2. Heartbeat is per-agent but should be per-claw-instance (global)
3. Scheduler only handles one agent — should manage all agents on this instance
4. wakeAgent used api.spawn (non-existent) → now uses dispatchInboundMessage (verified)
## Target Design
### Plugin State
```typescript
// Local schedule cache: { agentId → [slots] }
const schedules: Map<string, CalendarSlotResponse[]> = new Map();
```
### Sync Flow (every 5 min)
```
1. GET /calendar/sync?claw_identifier=xxx
- First call: server returns full { agentId → [slots] }
- Subsequent: server returns diff since last sync
2. Update local schedules map
3. Scan schedules for due slots:
for each agentId in schedules:
if has slot where scheduled_at <= now && status == not_started:
getAgentStatus(agentId, clawIdentifier) → busy?
if not busy → wakeAgent(agentId)
```
### Heartbeat (every 60s)
Simplified to liveness ping only:
```
POST /monitor/server/heartbeat
claw_identifier: xxx
→ server returns empty/ack
```
No slot data in heartbeat response.
### Wake Flow
```
dispatchInboundMessage:
SessionKey: agent:{agentId}:hf-wakeup
Body: "You have due slots. Follow the hf-wakeup workflow of skill hf-hangman-lab to proceed. Only reply WAKEUP_OK in this session."
Agent reads workflow → calls hf tools → sets own status to busy
```
### Agent ID Resolution
- **Sync**: agentId comes from server response (dict keys)
- **Wake**: agentId from local schedules dict key
- **Tool calls by agent**: agentId from tool ctx (same as padded-cell)
## Backend API Changes Needed
### New: GET /calendar/sync
```
GET /calendar/sync?claw_identifier=xxx
Headers: X-Claw-Identifier
Response (first call):
{
"full": true,
"schedules": {
"developer": [slot1, slot2, ...],
"operator": [slot3, ...]
},
"syncToken": "abc123"
}
Response (subsequent, with ?since=abc123):
{
"full": false,
"diff": [
{ "op": "add", "agent": "developer", "slot": {...} },
{ "op": "update", "agent": "developer", "slotId": 5, "patch": {...} },
{ "op": "remove", "agent": "operator", "slotId": 3 }
],
"syncToken": "def456"
}
```
### Existing: POST /calendar/agent/status
Keep as-is but ensure it accepts agentId + clawIdentifier as params:
```
POST /calendar/agent/status
{ agent_id, claw_identifier, status }
```
## Implementation Order
1. Backend: Add /calendar/sync endpoint
2. Plugin: Replace CalendarBridgeClient single-agent design with multi-agent
3. Plugin: Replace CalendarScheduler with new sync+check loop
4. Plugin: wakeAgent uses dispatchInboundMessage (done)
5. Plugin: Tool handlers get agentId from ctx (like padded-cell)
## Files to Change
### Backend (HarborForge.Backend)
- New route: `/calendar/sync`
- New service: schedule diff tracking per claw_identifier
### Plugin
- `plugin/calendar/calendar-bridge.ts` — remove agentId binding, add sync()
- `plugin/calendar/scheduler.ts` — rewrite to multi-agent sync+check
- `plugin/calendar/schedule-cache.ts` — already exists, adapt to multi-agent
- `plugin/index.ts` — update wakeAgent, getAgentStatus to accept agentId

View File

@@ -169,6 +169,74 @@ export class CalendarBridgeClient {
return this.sendBoolean('POST', url, body); return this.sendBoolean('POST', url, body);
} }
/**
* Fetch the full day schedule for this agent.
*
* Unlike heartbeat() which only returns pending (NOT_STARTED/DEFERRED) slots,
* this returns ALL slots for the given date, enabling the plugin to maintain
* a complete local view of today's schedule.
*
* @param date Date string in YYYY-MM-DD format
* @returns Array of all slots for the day, or null if unreachable
*/
async getDaySchedule(date: string): Promise<CalendarSlotResponse[] | null> {
const url = `${this.baseUrl}/calendar/day?date=${encodeURIComponent(date)}`;
try {
const response = await this.fetchJson<{ slots: CalendarSlotResponse[] }>(url, {
method: 'GET',
headers: {
'X-Agent-ID': this.config.agentId,
'X-Claw-Identifier': this.config.clawIdentifier,
},
});
return response?.slots ?? null;
} catch {
return null;
}
}
/**
* Sync today's schedules for all agents on this claw instance.
*
* Returns { agentId → slots[] } for all agents with matching claw_identifier.
* This is the primary data source for the multi-agent schedule cache.
*/
async syncSchedules(): Promise<{ schedules: Record<string, any[]>; date: string } | null> {
const url = `${this.baseUrl}/calendar/sync`;
try {
const response = await this.fetchJson<{ schedules: Record<string, any[]>; date: string }>(url, {
method: 'GET',
headers: {
'X-Claw-Identifier': this.config.clawIdentifier,
},
});
return response;
} catch {
return null;
}
}
/**
* Get a specific agent's status.
*
* @param agentId The agent ID to query
*/
async getAgentStatus(agentId: string): Promise<string | null> {
const url = `${this.baseUrl}/calendar/agent/status?agent_id=${encodeURIComponent(agentId)}`;
try {
const response = await this.fetchJson<{ status: string }>(url, {
method: 'GET',
headers: {
'X-Agent-ID': agentId,
'X-Claw-Identifier': this.config.clawIdentifier,
},
});
return response?.status ?? null;
} catch {
return null;
}
}
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------
// Internal helpers // Internal helpers
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------

View File

@@ -31,3 +31,4 @@
export * from './types'; export * from './types';
export * from './calendar-bridge'; export * from './calendar-bridge';
export * from './scheduler'; export * from './scheduler';
export * from './schedule-cache';

View File

@@ -0,0 +1,101 @@
/**
* Multi-agent local schedule cache.
*
* Maintains today's schedule for all agents on this claw instance.
* Synced periodically from HF backend via /calendar/sync endpoint.
*/
export interface CachedSlot {
id: number | null;
virtual_id: string | null;
slot_type: string;
estimated_duration: number;
scheduled_at: string;
status: string;
priority: number;
event_type: string | null;
event_data: Record<string, unknown> | null;
[key: string]: unknown;
}
export class MultiAgentScheduleCache {
/** { agentId → slots[] } */
private schedules: Map<string, CachedSlot[]> = new Map();
private lastSyncAt: Date | null = null;
private cachedDate: string | null = null;
/**
* Replace cache with data from /calendar/sync response.
*/
sync(date: string, schedules: Record<string, CachedSlot[]>): void {
if (this.cachedDate !== date) {
this.schedules.clear();
}
this.cachedDate = date;
for (const [agentId, slots] of Object.entries(schedules)) {
this.schedules.set(agentId, slots);
}
this.lastSyncAt = new Date();
}
/**
* Get agents that have due (overdue or current) slots.
* Returns [agentId, dueSlots[]] pairs.
*/
getAgentsWithDueSlots(now: Date): Array<{ agentId: string; slots: CachedSlot[] }> {
const results: Array<{ agentId: string; slots: CachedSlot[] }> = [];
for (const [agentId, slots] of this.schedules) {
const due = slots.filter((s) => {
if (s.status !== 'not_started' && s.status !== 'deferred') return false;
const scheduledAt = this.parseScheduledTime(s.scheduled_at);
return scheduledAt !== null && scheduledAt <= now;
});
if (due.length > 0) {
// Sort by priority descending
due.sort((a, b) => (b.priority ?? 0) - (a.priority ?? 0));
results.push({ agentId, slots: due });
}
}
return results;
}
/**
* Get all agent IDs in the cache.
*/
getAgentIds(): string[] {
return Array.from(this.schedules.keys());
}
/**
* Get slots for a specific agent.
*/
getAgentSlots(agentId: string): CachedSlot[] {
return this.schedules.get(agentId) ?? [];
}
/**
* Get cache status for debugging.
*/
getStatus(): { agentCount: number; totalSlots: number; lastSyncAt: string | null; cachedDate: string | null } {
let totalSlots = 0;
for (const slots of this.schedules.values()) totalSlots += slots.length;
return {
agentCount: this.schedules.size,
totalSlots,
lastSyncAt: this.lastSyncAt?.toISOString() ?? null,
cachedDate: this.cachedDate,
};
}
private parseScheduledTime(scheduledAt: string): Date | null {
if (/^\d{2}:\d{2}(:\d{2})?$/.test(scheduledAt)) {
if (!this.cachedDate) return null;
return new Date(`${this.cachedDate}T${scheduledAt}Z`);
}
const d = new Date(scheduledAt);
return isNaN(d.getTime()) ? null : d;
}
}

View File

@@ -19,6 +19,7 @@
import { writeFileSync, readFileSync, existsSync, mkdirSync } from 'fs'; import { writeFileSync, readFileSync, existsSync, mkdirSync } from 'fs';
import { join, dirname } from 'path'; import { join, dirname } from 'path';
import { CalendarBridgeClient } from './calendar-bridge'; import { CalendarBridgeClient } from './calendar-bridge';
import { ScheduleCache } from './schedule-cache';
import { import {
CalendarSlotResponse, CalendarSlotResponse,
SlotStatus, SlotStatus,
@@ -44,6 +45,8 @@ export interface CalendarSchedulerConfig {
}; };
/** Heartbeat interval in milliseconds (default: 60000) */ /** Heartbeat interval in milliseconds (default: 60000) */
heartbeatIntervalMs?: number; heartbeatIntervalMs?: number;
/** Schedule sync interval in milliseconds (default: 300000 = 5 min) */
syncIntervalMs?: number;
/** Enable verbose debug logging */ /** Enable verbose debug logging */
debug?: boolean; debug?: boolean;
/** Directory for state persistence (default: plugin data dir) */ /** Directory for state persistence (default: plugin data dir) */
@@ -95,8 +98,10 @@ interface SchedulerState {
currentSlot: CalendarSlotResponse | null; currentSlot: CalendarSlotResponse | null;
/** Last heartbeat timestamp */ /** Last heartbeat timestamp */
lastHeartbeatAt: Date | null; lastHeartbeatAt: Date | null;
/** Interval handle for cleanup */ /** Heartbeat interval handle */
intervalHandle: ReturnType<typeof setInterval> | null; intervalHandle: ReturnType<typeof setInterval> | null;
/** Schedule sync interval handle */
syncIntervalHandle: ReturnType<typeof setInterval> | null;
/** Set of slot IDs that have been deferred in current session */ /** Set of slot IDs that have been deferred in current session */
deferredSlotIds: Set<string>; deferredSlotIds: Set<string>;
/** Whether agent is currently processing a slot */ /** Whether agent is currently processing a slot */
@@ -117,10 +122,13 @@ export class CalendarScheduler {
private config: Required<CalendarSchedulerConfig>; private config: Required<CalendarSchedulerConfig>;
private state: SchedulerState; private state: SchedulerState;
private stateFilePath: string; private stateFilePath: string;
/** Local cache of today's full schedule, synced periodically from backend */
private scheduleCache: ScheduleCache = new ScheduleCache();
constructor(config: CalendarSchedulerConfig) { constructor(config: CalendarSchedulerConfig) {
this.config = { this.config = {
heartbeatIntervalMs: 60000, // 1 minute default heartbeatIntervalMs: 60000, // 1 minute default
syncIntervalMs: 300_000, // 5 minutes default
debug: false, debug: false,
stateDir: this.getDefaultStateDir(), stateDir: this.getDefaultStateDir(),
...config, ...config,
@@ -133,6 +141,7 @@ export class CalendarScheduler {
currentSlot: null, currentSlot: null,
lastHeartbeatAt: null, lastHeartbeatAt: null,
intervalHandle: null, intervalHandle: null,
syncIntervalHandle: null,
deferredSlotIds: new Set(), deferredSlotIds: new Set(),
isProcessing: false, isProcessing: false,
isRestartPending: false, isRestartPending: false,
@@ -327,14 +336,21 @@ export class CalendarScheduler {
this.state.isRestartPending = false; this.state.isRestartPending = false;
this.config.logger.info('Calendar scheduler started'); this.config.logger.info('Calendar scheduler started');
// Run initial heartbeat immediately // Run initial sync + heartbeat immediately
this.runSync();
this.runHeartbeat(); this.runHeartbeat();
// Schedule periodic heartbeats // Schedule periodic heartbeats (slot execution checks)
this.state.intervalHandle = setInterval( this.state.intervalHandle = setInterval(
() => this.runHeartbeat(), () => this.runHeartbeat(),
this.config.heartbeatIntervalMs this.config.heartbeatIntervalMs
); );
// Schedule periodic schedule sync (full day schedule refresh)
this.state.syncIntervalHandle = setInterval(
() => this.runSync(),
this.config.syncIntervalMs
);
} }
/** /**
@@ -348,10 +364,41 @@ export class CalendarScheduler {
clearInterval(this.state.intervalHandle); clearInterval(this.state.intervalHandle);
this.state.intervalHandle = null; this.state.intervalHandle = null;
} }
if (this.state.syncIntervalHandle) {
clearInterval(this.state.syncIntervalHandle);
this.state.syncIntervalHandle = null;
}
this.config.logger.info('Calendar scheduler stopped'); this.config.logger.info('Calendar scheduler stopped');
} }
/**
* Sync today's full schedule from backend into local cache.
* Runs every syncIntervalMs (default: 5 min).
* Catches new slots assigned by other agents or the manager.
*/
async runSync(): Promise<void> {
if (!this.state.isRunning || this.state.isRestartPending) return;
const today = new Date().toISOString().slice(0, 10);
try {
const slots = await this.config.bridge.getDaySchedule(today);
if (slots) {
this.scheduleCache.sync(today, slots);
this.logDebug(`Schedule synced: ${slots.length} slots for ${today}`);
}
} catch (err) {
this.config.logger.warn(`Schedule sync failed: ${String(err)}`);
}
}
/**
* Get the local schedule cache (for status reporting / tools).
*/
getScheduleCache(): ScheduleCache {
return this.scheduleCache;
}
/** /**
* Execute a single heartbeat cycle. * Execute a single heartbeat cycle.
* Fetches pending slots and handles execution logic. * Fetches pending slots and handles execution logic.
@@ -611,13 +658,11 @@ Task Code: ${code}
Estimated Duration: ${duration} minutes Estimated Duration: ${duration} minutes
Slot Type: ${slot.slot_type} Slot Type: ${slot.slot_type}
Priority: ${slot.priority} Priority: ${slot.priority}
Working Sessions: ${jobData.working_sessions?.join(', ') || 'none recorded'}
Please focus on this task for the allocated time. When you finish or need to pause, Follow the daily-routine skill's task-handson workflow to execute this task.
report your progress back to the calendar system. Use harborforge_calendar_complete when finished, or harborforge_calendar_pause to pause.
Before going idle, check for overdue slots as described in the slot-complete workflow.`;
Working sessions: ${jobData.working_sessions?.join(', ') || 'none recorded'}
Start working on ${code} now.`;
} }
/** /**
@@ -630,19 +675,15 @@ Start working on ${code} now.`;
switch (sysData.event) { switch (sysData.event) {
case 'ScheduleToday': case 'ScheduleToday':
return `System Event: Schedule Today return `System Event: Schedule Today
Please review today's calendar and schedule any pending tasks or planning activities.
Estimated time: ${slot.estimated_duration} minutes. Estimated time: ${slot.estimated_duration} minutes.
Check your calendar and plan the day's work.`; Follow the daily-routine skill's plan-schedule workflow to plan today's work.`;
case 'SummaryToday': case 'SummaryToday':
return `System Event: Daily Summary return `System Event: Daily Summary
Please provide a summary of today's activities and progress.
Estimated time: ${slot.estimated_duration} minutes. Estimated time: ${slot.estimated_duration} minutes.
Review what was accomplished and prepare end-of-day notes.`; Review today's completed, deferred, and abandoned slots. Write a summary to your daily note (memory/YYYY-MM-DD.md).`;
case 'ScheduledGatewayRestart': case 'ScheduledGatewayRestart':
return `System Event: Scheduled Gateway Restart return `System Event: Scheduled Gateway Restart

View File

@@ -68,6 +68,13 @@ export default {
return getPluginConfig(api); return getPluginConfig(api);
} }
/** Resolve agent ID from env, config, or fallback. */
function resolveAgentId(): string {
if (process.env.AGENT_ID) return process.env.AGENT_ID;
const cfg = api.runtime?.config?.loadConfig?.();
return cfg?.agents?.list?.[0]?.id ?? cfg?.agents?.defaults?.id ?? 'unknown';
}
/** /**
* Get the monitor bridge client if monitor_port is configured. * Get the monitor bridge client if monitor_port is configured.
*/ */
@@ -168,7 +175,7 @@ export default {
// Fallback: query backend for agent status // Fallback: query backend for agent status
const live = resolveConfig(); const live = resolveConfig();
const agentId = process.env.AGENT_ID || 'unknown'; const agentId = resolveAgentId();
try { try {
const response = await fetch(`${live.backendUrl}/calendar/agent/status?agent_id=${agentId}`, { const response = await fetch(`${live.backendUrl}/calendar/agent/status?agent_id=${agentId}`, {
headers: { headers: {
@@ -188,56 +195,51 @@ export default {
} }
/** /**
* Wake/spawn agent with task context for slot execution. * Wake agent via dispatchInboundMessage — same mechanism used by Discord plugin.
* This is the callback invoked by CalendarScheduler when a slot is ready. * Direct in-process call, no WebSocket or CLI needed.
*/ */
async function wakeAgent(context: AgentWakeContext): Promise<boolean> { async function wakeAgent(agentId: string): Promise<boolean> {
logger.info(`Waking agent for slot: ${context.taskDescription}`); logger.info(`Waking agent ${agentId}: has due slots`);
const sessionKey = `agent:${agentId}:hf-wakeup`;
try { try {
// Method 1: Use OpenClaw spawn API if available (preferred) const sdkPath = 'openclaw/plugin-sdk/reply-runtime';
if (api.spawn) { const { dispatchInboundMessageWithDispatcher } = await import(
const result = await api.spawn({ /* webpackIgnore: true */ sdkPath
task: context.prompt, );
timeoutSeconds: context.slot.estimated_duration * 60, // Convert to seconds
});
if (result?.sessionId) { const cfg = api.runtime?.config?.loadConfig?.();
logger.info(`Agent spawned for calendar slot: session=${result.sessionId}`); if (!cfg) {
logger.error('Cannot load OpenClaw config for dispatch');
// Track session completion return false;
trackSessionCompletion(result.sessionId, context);
return true;
}
} }
// Method 2: Send notification/alert to wake agent (fallback) const wakeupMessage = `You have due slots. Follow the \`hf-wakeup\` workflow of skill \`hf-hangman-lab\` to proceed. Only reply \`WAKEUP_OK\` in this session.`;
// This relies on the agent's heartbeat to check for notifications
logger.warn('OpenClaw spawn API not available, using notification fallback');
// Send calendar wakeup notification via backend const result = await dispatchInboundMessageWithDispatcher({
const live = resolveConfig(); ctx: {
const agentId = process.env.AGENT_ID || 'unknown'; Body: wakeupMessage,
SessionKey: sessionKey,
const notifyResponse = await fetch(`${live.backendUrl}/calendar/agent/notify`, { From: 'harborforge-calendar',
method: 'POST', Provider: 'harborforge',
headers: { },
'Content-Type': 'application/json', cfg,
'X-Agent-ID': agentId, dispatcherOptions: {
'X-Claw-Identifier': live.identifier || hostname(), deliver: async (payload: any) => {
const text = (payload.text || '').trim();
logger.info(`Agent ${agentId} wakeup reply: ${text.slice(0, 100)}`);
},
}, },
body: JSON.stringify({
agent_id: agentId,
message: context.prompt,
slot_id: context.slot.id || context.slot.virtual_id,
task_description: context.taskDescription,
}),
}); });
return notifyResponse.ok; logger.info(`Agent ${agentId} dispatched: ${result?.status || 'ok'}`);
return true;
} catch (err) { } catch (err: any) {
logger.error('Failed to wake agent:', err); const msg = err?.message || err?.code || String(err);
const stack = err?.stack?.split('\n').slice(0, 3).join(' | ') || '';
logger.error(`Failed to dispatch agent for slot: ${msg} ${stack}`);
return false; return false;
} }
} }
@@ -279,27 +281,69 @@ export default {
*/ */
function startCalendarScheduler(): void { function startCalendarScheduler(): void {
const live = resolveConfig(); const live = resolveConfig();
const agentId = process.env.AGENT_ID || 'unknown';
// Create calendar bridge client // Create bridge client (claw-instance level, not per-agent)
const calendarBridge = createCalendarBridgeClient( const calendarBridge = createCalendarBridgeClient(
api, api,
live.backendUrl || 'https://monitor.hangman-lab.top', live.backendUrl || 'https://monitor.hangman-lab.top',
agentId 'unused' // agentId no longer needed at bridge level
); );
// Create and start scheduler // Multi-agent sync + check loop
calendarScheduler = createCalendarScheduler({ const { MultiAgentScheduleCache } = require('./calendar/schedule-cache') as typeof import('./calendar/schedule-cache');
bridge: calendarBridge, const scheduleCache = new MultiAgentScheduleCache();
getAgentStatus,
wakeAgent,
logger,
heartbeatIntervalMs: 60000, // 1 minute
debug: live.logLevel === 'debug',
});
calendarScheduler.start(); const SYNC_INTERVAL_MS = 300_000; // 5 min
logger.info('Calendar scheduler started'); const CHECK_INTERVAL_MS = 30_000; // 30 sec
// Sync: pull all agent schedules from backend
async function runSync() {
try {
const result = await calendarBridge.syncSchedules();
if (result) {
scheduleCache.sync(result.date, result.schedules);
const status = scheduleCache.getStatus();
logger.info(`Schedule synced: ${status.agentCount} agents, ${status.totalSlots} slots`);
}
} catch (err) {
logger.warn(`Schedule sync failed: ${String(err)}`);
}
}
// Check: find agents with due slots and wake them
async function runCheck() {
const now = new Date();
const agentsWithDue = scheduleCache.getAgentsWithDueSlots(now);
for (const { agentId } of agentsWithDue) {
// Check if agent is busy
const status = await calendarBridge.getAgentStatus(agentId);
if (status === 'busy' || status === 'offline' || status === 'exhausted') {
continue;
}
// Wake the agent
await wakeAgent(agentId);
}
}
// Initial sync
runSync();
// Start intervals
const syncHandle = setInterval(runSync, SYNC_INTERVAL_MS);
const checkHandle = setInterval(runCheck, CHECK_INTERVAL_MS);
// Store handles for cleanup (reuse calendarScheduler variable)
(calendarScheduler as any) = {
stop() {
clearInterval(syncHandle);
clearInterval(checkHandle);
logger.info('Calendar scheduler stopped');
},
};
logger.info('Calendar scheduler started (multi-agent sync mode)');
} }
/** /**