Compare commits

...

4 Commits

Author SHA1 Message Date
operator
ec09578de3 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>
2026-04-19 09:32:36 +00:00
operator
248adfaafd fix: use runtime API for version and agent list instead of subprocess
Use api.runtime.version for openclaw version and
api.runtime.config.loadConfig() for agent list. Eliminates the
periodic openclaw agents list subprocess that caused high CPU usage.
2026-04-16 15:53:20 +00:00
operator
e4ac7b7af3 fix: disable periodic openclaw agents list subprocess
Spawning a full openclaw CLI process every 30s to list agents is too
heavy — each invocation loads all plugins (~16s) and hangs until killed.
Return empty array for now until a lighter mechanism is available.
2026-04-16 15:26:55 +00:00
operator
2088cd12b4 fix: use OPENCLAW_SERVICE_VERSION for real version and increase agent list timeout
api.version returns plugin API version (0.2.0), not the openclaw release
version. Use OPENCLAW_SERVICE_VERSION env var set by the gateway instead.
Also increase listOpenClawAgents timeout from 15s to 30s since plugin
loading takes ~16s on T2.
2026-04-16 15:12:35 +00:00
7 changed files with 477 additions and 124 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

@@ -1,8 +1,3 @@
import { execFile } from 'child_process';
import { promisify } from 'util';
const execFileAsync = promisify(execFile);
export interface OpenClawAgentInfo { export interface OpenClawAgentInfo {
name: string; name: string;
isDefault?: boolean; isDefault?: boolean;
@@ -14,70 +9,38 @@ export interface OpenClawAgentInfo {
routing?: string; routing?: string;
} }
export async function listOpenClawAgents(logger?: { debug?: (...args: any[]) => void; warn?: (...args: any[]) => void }): Promise<OpenClawAgentInfo[]> { export async function listOpenClawAgents(_logger?: { debug?: (...args: any[]) => void; warn?: (...args: any[]) => void }): Promise<OpenClawAgentInfo[]> {
try {
const { stdout } = await execFileAsync('openclaw', ['agents', 'list'], {
timeout: 15000,
maxBuffer: 1024 * 1024,
});
return parseOpenClawAgents(stdout);
} catch (err) {
logger?.warn?.('Failed to run `openclaw agents list`', err);
return []; return [];
} }
}
export function parseOpenClawAgents(text: string): OpenClawAgentInfo[] { export function parseOpenClawAgents(text: string): OpenClawAgentInfo[] {
const lines = text.split(/\r?\n/); const lines = text.split(/\r?\n/);
const out: OpenClawAgentInfo[] = []; const out: OpenClawAgentInfo[] = [];
let current: OpenClawAgentInfo | null = null; let current: OpenClawAgentInfo | null = null;
const push = () => { if (current) out.push(current); current = null; };
const push = () => {
if (current) out.push(current);
current = null;
};
for (const raw of lines) { for (const raw of lines) {
const line = raw.trimEnd(); const line = raw.trimEnd();
if (!line.trim() || line.startsWith('Agents:') || line.startsWith('Routing rules map') || line.startsWith('Channel status reflects')) continue; if (!line.trim() || line.startsWith("Agents:") || line.startsWith("Routing rules map") || line.startsWith("Channel status reflects")) continue;
if (line.startsWith('- ')) { if (line.startsWith("- ")) {
push(); push();
const m = line.match(/^-\s+(.+?)(?:\s+\((default)\))?$/); const m = line.match(/^-\s+(.+?)(?:\s+\((default)\))?$/);
current = { current = { name: m?.[1] || line.slice(2).trim(), isDefault: m?.[2] === "default" };
name: m?.[1] || line.slice(2).trim(),
isDefault: m?.[2] === 'default',
};
continue; continue;
} }
if (!current) continue; if (!current) continue;
const trimmed = line.trim(); const trimmed = line.trim();
const idx = trimmed.indexOf(':'); const idx = trimmed.indexOf(":");
if (idx === -1) continue; if (idx === -1) continue;
const key = trimmed.slice(0, idx).trim(); const key = trimmed.slice(0, idx).trim();
const value = trimmed.slice(idx + 1).trim(); const value = trimmed.slice(idx + 1).trim();
switch (key) { switch (key) {
case 'Identity': case "Identity": current.identity = value; break;
current.identity = value; case "Workspace": current.workspace = value; break;
break; case "Agent dir": current.agentDir = value; break;
case 'Workspace': case "Model": current.model = value; break;
current.workspace = value; case "Routing rules": { const n = Number(value); current.routingRules = Number.isFinite(n) ? n : undefined; break; }
break; case "Routing": current.routing = value; break;
case 'Agent dir': default: break;
current.agentDir = value;
break;
case 'Model':
current.model = value;
break;
case 'Routing rules': {
const n = Number(value);
current.routingRules = Number.isFinite(n) ? n : undefined;
break;
}
case 'Routing':
current.routing = value;
break;
default:
break;
} }
} }
push(); push();

View File

@@ -14,7 +14,7 @@
import { hostname, freemem, totalmem, uptime, loadavg, platform } from 'os'; import { hostname, freemem, totalmem, uptime, loadavg, platform } from 'os';
import { getPluginConfig } from './core/config'; import { getPluginConfig } from './core/config';
import { MonitorBridgeClient, type OpenClawMeta } from './core/monitor-bridge'; import { MonitorBridgeClient, type OpenClawMeta } from './core/monitor-bridge';
import { listOpenClawAgents } from './core/openclaw-agents'; import type { OpenClawAgentInfo } from './core/openclaw-agents';
import { registerGatewayStartHook } from './hooks/gateway-start'; import { registerGatewayStartHook } from './hooks/gateway-start';
import { registerGatewayStopHook } from './hooks/gateway-stop'; import { registerGatewayStopHook } from './hooks/gateway-stop';
import { import {
@@ -32,6 +32,12 @@ interface PluginAPI {
warn: (...args: any[]) => void; warn: (...args: any[]) => void;
}; };
version?: string; version?: string;
runtime?: {
version?: string;
config?: {
loadConfig?: () => any;
};
};
config?: Record<string, unknown>; config?: Record<string, unknown>;
pluginConfig?: Record<string, unknown>; pluginConfig?: Record<string, unknown>;
on: (event: string, handler: () => void) => void; on: (event: string, handler: () => void) => void;
@@ -62,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.
*/ */
@@ -96,7 +109,7 @@ export default {
avg15: load[2], avg15: load[2],
}, },
openclaw: { openclaw: {
version: api.version || 'unknown', version: api.runtime?.version || api.version || 'unknown',
pluginVersion: '0.3.1', // Bumped for PLG-CAL-004 pluginVersion: '0.3.1', // Bumped for PLG-CAL-004
}, },
timestamp: new Date().toISOString(), timestamp: new Date().toISOString(),
@@ -118,10 +131,21 @@ export default {
const bridgeClient = getBridgeClient(); const bridgeClient = getBridgeClient();
if (!bridgeClient) return; if (!bridgeClient) return;
let agentNames: string[] = [];
try {
const cfg = api.runtime?.config?.loadConfig?.();
const agentsList = cfg?.agents?.list;
if (Array.isArray(agentsList)) {
agentNames = agentsList
.map((a: any) => typeof a === 'string' ? a : a?.name)
.filter(Boolean);
}
} catch { /* non-fatal */ }
const meta: OpenClawMeta = { const meta: OpenClawMeta = {
version: api.version || 'unknown', version: api.runtime?.version || api.version || 'unknown',
plugin_version: '0.3.1', plugin_version: '0.3.1',
agents: await listOpenClawAgents(logger), agents: agentNames.map(name => ({ name })),
}; };
const ok = await bridgeClient.pushOpenClawMeta(meta); const ok = await bridgeClient.pushOpenClawMeta(meta);
@@ -151,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: {
@@ -171,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;
} }
} }
@@ -262,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)');
} }
/** /**