Compare commits

..

4 Commits

Author SHA1 Message Date
operator
05bf51930d Revert "feat: Discord-based agent wakeup replacing spawn"
This reverts commit be30b4b3f4.
2026-04-18 20:45:37 +00:00
operator
be30b4b3f4 feat: Discord-based agent wakeup replacing spawn
New wakeup flow:
1. Create private Discord channel for the agent
2. Send wakeup message with slot context + workflow reference
3. If Dirigent detected (globalThis.__dirigent), create work-type channel
4. Fallback to api.spawn if Discord not configured

New config fields: discordBotToken, discordGuildId
New file: plugin/calendar/discord-wakeup.ts

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-18 20:28:59 +00:00
operator
7e4750fcc4 feat: add local schedule cache + periodic sync to CalendarScheduler
- New ScheduleCache class: maintains today's full schedule locally
- CalendarBridgeClient.getDaySchedule(): fetch all slots for a date
- Scheduler now runs two intervals:
  - Heartbeat (60s): existing slot execution flow (unchanged)
  - Sync (5min): pulls full day schedule into local cache
- Exposes getScheduleCache() for tools and status reporting

This enables the plugin to detect slots assigned by other agents
between heartbeats and provides a complete local view of the schedule.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-18 17:45:32 +00:00
operator
6d59741086 feat: align wakeup prompts with daily-routine skill workflows
Update CalendarScheduler prompt templates to reference daily-routine
skill workflows (task-handson, plan-schedule, slot-complete) instead
of generic instructions.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-18 17:31:36 +00:00
5 changed files with 261 additions and 325 deletions

View File

@@ -1,118 +1,112 @@
# CalendarScheduler Refactor Plan (v2) # CalendarScheduler Refactor Plan
> Updated 2026-04-19 based on architecture discussion with hang ## Current Design
## Current Issues ```
Every 60s:
heartbeat() → POST /calendar/agent/heartbeat → returns pending slots
if idle → select highest priority → executeSlot → wakeAgent(spawn)
if busy → defer all pending slots
```
1. `process.env.AGENT_ID` doesn't exist in plugins subprocess — always 'unknown' **Problems:**
2. Heartbeat is per-agent but should be per-claw-instance (global) 1. Every heartbeat queries backend for pending slots — no local awareness of full schedule
3. Scheduler only handles one agent — should manage all agents on this instance 2. Cannot detect slots assigned by other agents between heartbeats
4. wakeAgent used api.spawn (non-existent) → now uses dispatchInboundMessage (verified) 3. 60s interval is too frequent for sync but too infrequent for precise wakeup
4. Wakeup via `api.spawn()` creates a plain session, not a Discord private channel
## Target Design ## Target Design
### Plugin State ```
Every 5-10 min (sync interval):
syncSchedule() → GET /calendar/day → update local today cache
Every 30s (check interval):
checkDueSlots() → scan local cache for due slots
if due slot found:
confirmAgentStatus() → GET /calendar/agent/status
if not busy → wakeAgent (via Dirigent moderator bot private channel)
```
## Changes Required
### 1. Add Local Schedule Cache
New class `ScheduleCache`:
```typescript
class ScheduleCache {
private slots: Map<string, CalendarSlotResponse>; // slotId → slot
private lastSyncAt: Date | null;
async sync(bridge: CalendarBridgeClient): Promise<void>; // fetch today's full schedule
getDueSlots(now: Date): CalendarSlotResponse[]; // scheduled_at <= now && NOT_STARTED/DEFERRED
updateSlot(id: string, update: Partial<CalendarSlotResponse>): void; // local update
getAll(): CalendarSlotResponse[];
}
```
### 2. Add CalendarBridgeClient.getDaySchedule()
New endpoint call:
```typescript
async getDaySchedule(date: string): Promise<CalendarSlotResponse[]>
// GET /calendar/day?date=YYYY-MM-DD
```
This fetches ALL slots for the day, not just pending ones. The existing `heartbeat()` only returns NOT_STARTED/DEFERRED.
### 3. Split Heartbeat into Sync + Check
**Replace** single `runHeartbeat()` with two intervals:
```typescript ```typescript
// Local schedule cache: { agentId → [slots] } // Sync: every 5 min — pull full schedule from backend
const schedules: Map<string, CalendarSlotResponse[]> = new Map(); this.syncInterval = setInterval(() => this.runSync(), 300_000);
// Check: every 30s — scan local cache for due slots
this.checkInterval = setInterval(() => this.runCheck(), 30_000);
``` ```
### Sync Flow (every 5 min) `runSync()`:
1. `bridge.getDaySchedule(today)` → update cache
2. Still send heartbeat to keep backend informed of agent liveness
``` `runCheck()`:
1. GET /calendar/sync?claw_identifier=xxx 1. `cache.getDueSlots(now)` → find due slots
- First call: server returns full { agentId → [slots] } 2. Filter out session-deferred slots
- Subsequent: server returns diff since last sync 3. If agent idle → select highest priority → execute
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) ### 4. Wakeup via Dirigent (future)
Simplified to liveness ping only: Change `wakeAgent()` to create a private Discord channel via Dirigent moderator bot instead of `api.spawn()`. This requires:
``` - Access to Dirigent's moderator bot token or cross-plugin API
POST /monitor/server/heartbeat - Creating a private channel with only the target agent
claw_identifier: xxx - Posting the wakeup prompt as a message
→ server returns empty/ack
```
No slot data in heartbeat response. **For now:** Keep `api.spawn()` as the wakeup method. The Dirigent integration can be added later as it requires cross-plugin coordination.
### 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 ## Implementation Order
1. Backend: Add /calendar/sync endpoint 1. Add `ScheduleCache` class (new file: `plugin/calendar/schedule-cache.ts`)
2. Plugin: Replace CalendarBridgeClient single-agent design with multi-agent 2. Add `getDaySchedule()` to `CalendarBridgeClient`
3. Plugin: Replace CalendarScheduler with new sync+check loop 3. Refactor `CalendarScheduler`:
4. Plugin: wakeAgent uses dispatchInboundMessage (done) - Replace single interval with sync + check intervals
5. Plugin: Tool handlers get agentId from ctx (like padded-cell) - Use cache instead of heartbeat for slot discovery
- Keep heartbeat for agent liveness reporting (reduced frequency)
4. Update state persistence for new structure
5. Keep existing wakeAgent/completion/abort/pause/resume tools unchanged
## Files to Change ## Files to Modify
### Backend (HarborForge.Backend) | File | Changes |
- New route: `/calendar/sync` |------|---------|
- New service: schedule diff tracking per claw_identifier | `plugin/calendar/schedule-cache.ts` | New file |
| `plugin/calendar/calendar-bridge.ts` | Add `getDaySchedule()` |
| `plugin/calendar/scheduler.ts` | Refactor heartbeat → sync + check |
| `plugin/calendar/index.ts` | Export new types |
### Plugin ## Risk Assessment
- `plugin/calendar/calendar-bridge.ts` — remove agentId binding, add sync()
- `plugin/calendar/scheduler.ts` — rewrite to multi-agent sync+check - **Low risk:** ScheduleCache is additive, doesn't break existing behavior
- `plugin/calendar/schedule-cache.ts` — already exists, adapt to multi-agent - **Medium risk:** Splitting heartbeat changes core scheduling logic
- `plugin/index.ts` — update wakeAgent, getAgentStatus to accept agentId - **Mitigation:** Keep `heartbeat()` method intact, use it for liveness reporting alongside new sync

View File

@@ -195,48 +195,6 @@ export class CalendarBridgeClient {
} }
} }
/**
* 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

@@ -1,97 +1,105 @@
/** /**
* Multi-agent local schedule cache. * Local cache of today's calendar schedule.
* * Synced periodically from HF backend, checked locally for due slots.
* Maintains today's schedule for all agents on this claw instance.
* Synced periodically from HF backend via /calendar/sync endpoint.
*/ */
export interface CachedSlot { import type { CalendarSlotResponse } from "./types.js";
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 { export class ScheduleCache {
/** { agentId → slots[] } */ private slots: Map<string, CalendarSlotResponse> = new Map();
private schedules: Map<string, CachedSlot[]> = new Map();
private lastSyncAt: Date | null = null; private lastSyncAt: Date | null = null;
private cachedDate: string | null = null; private cachedDate: string | null = null; // YYYY-MM-DD
/** /**
* Replace cache with data from /calendar/sync response. * Replace the cache with a fresh schedule from backend.
*/ */
sync(date: string, schedules: Record<string, CachedSlot[]>): void { sync(date: string, slots: CalendarSlotResponse[]): void {
// If date changed, clear old data
if (this.cachedDate !== date) { if (this.cachedDate !== date) {
this.schedules.clear(); this.slots.clear();
} }
this.cachedDate = date; this.cachedDate = date;
for (const [agentId, slots] of Object.entries(schedules)) { // Merge: update existing slots, add new ones
this.schedules.set(agentId, slots); const incomingIds = new Set<string>();
for (const slot of slots) {
const id = this.getSlotId(slot);
incomingIds.add(id);
this.slots.set(id, slot);
} }
// Remove slots that no longer exist on backend (cancelled etc.)
for (const id of this.slots.keys()) {
if (!incomingIds.has(id)) {
this.slots.delete(id);
}
}
this.lastSyncAt = new Date(); this.lastSyncAt = new Date();
} }
/** /**
* Get agents that have due (overdue or current) slots. * Get slots that are due (scheduled_at <= now) and still pending.
* Returns [agentId, dueSlots[]] pairs.
*/ */
getAgentsWithDueSlots(now: Date): Array<{ agentId: string; slots: CachedSlot[] }> { getDueSlots(now: Date): CalendarSlotResponse[] {
const results: Array<{ agentId: string; slots: CachedSlot[] }> = []; const results: CalendarSlotResponse[] = [];
for (const slot of this.slots.values()) {
if (slot.status !== "not_started" && slot.status !== "deferred") continue;
if (!slot.scheduled_at) continue;
for (const [agentId, slots] of this.schedules) { const scheduledAt = this.parseScheduledTime(slot.scheduled_at);
const due = slots.filter((s) => { if (scheduledAt && scheduledAt <= now) {
if (s.status !== 'not_started' && s.status !== 'deferred') return false; results.push(slot);
const scheduledAt = this.parseScheduledTime(s.scheduled_at); }
return scheduledAt !== null && scheduledAt <= now; }
});
if (due.length > 0) {
// Sort by priority descending // Sort by priority descending
due.sort((a, b) => (b.priority ?? 0) - (a.priority ?? 0)); results.sort((a, b) => (b.priority ?? 0) - (a.priority ?? 0));
results.push({ agentId, slots: due });
}
}
return results; return results;
} }
/** /**
* Get all agent IDs in the cache. * Update a slot locally (e.g., after status change).
*/ */
getAgentIds(): string[] { updateSlot(slotId: string, update: Partial<CalendarSlotResponse>): void {
return Array.from(this.schedules.keys()); const existing = this.slots.get(slotId);
if (existing) {
this.slots.set(slotId, { ...existing, ...update });
}
} }
/** /**
* Get slots for a specific agent. * Remove a slot from cache.
*/ */
getAgentSlots(agentId: string): CachedSlot[] { removeSlot(slotId: string): void {
return this.schedules.get(agentId) ?? []; this.slots.delete(slotId);
} }
/** /**
* Get cache status for debugging. * Get all cached slots.
*/ */
getStatus(): { agentCount: number; totalSlots: number; lastSyncAt: string | null; cachedDate: string | null } { getAll(): CalendarSlotResponse[] {
let totalSlots = 0; return Array.from(this.slots.values());
for (const slots of this.schedules.values()) totalSlots += slots.length; }
/**
* Get cache metadata.
*/
getStatus(): { slotCount: number; lastSyncAt: string | null; cachedDate: string | null } {
return { return {
agentCount: this.schedules.size, slotCount: this.slots.size,
totalSlots,
lastSyncAt: this.lastSyncAt?.toISOString() ?? null, lastSyncAt: this.lastSyncAt?.toISOString() ?? null,
cachedDate: this.cachedDate, cachedDate: this.cachedDate,
}; };
} }
private getSlotId(slot: CalendarSlotResponse): string {
return slot.virtual_id ?? String(slot.id);
}
private parseScheduledTime(scheduledAt: string): Date | null { private parseScheduledTime(scheduledAt: string): Date | null {
// scheduled_at can be "HH:MM:SS" (time only) or full ISO
if (/^\d{2}:\d{2}(:\d{2})?$/.test(scheduledAt)) { if (/^\d{2}:\d{2}(:\d{2})?$/.test(scheduledAt)) {
// Time-only: combine with cached date
if (!this.cachedDate) return null; if (!this.cachedDate) return null;
return new Date(`${this.cachedDate}T${scheduledAt}Z`); return new Date(`${this.cachedDate}T${scheduledAt}Z`);
} }

View File

@@ -1,3 +1,8 @@
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;
@@ -9,38 +14,70 @@ 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 = { name: m?.[1] || line.slice(2).trim(), isDefault: m?.[2] === "default" }; current = {
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": current.identity = value; break; case 'Identity':
case "Workspace": current.workspace = value; break; current.identity = value;
case "Agent dir": current.agentDir = value; break; break;
case "Model": current.model = value; break; case 'Workspace':
case "Routing rules": { const n = Number(value); current.routingRules = Number.isFinite(n) ? n : undefined; break; } current.workspace = value;
case "Routing": current.routing = value; break; break;
default: break; case 'Agent dir':
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 type { OpenClawAgentInfo } from './core/openclaw-agents'; import { listOpenClawAgents } 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,12 +32,6 @@ 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;
@@ -68,13 +62,6 @@ 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.
*/ */
@@ -109,7 +96,7 @@ export default {
avg15: load[2], avg15: load[2],
}, },
openclaw: { openclaw: {
version: api.runtime?.version || api.version || 'unknown', 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(),
@@ -131,21 +118,10 @@ 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.runtime?.version || api.version || 'unknown', version: api.version || 'unknown',
plugin_version: '0.3.1', plugin_version: '0.3.1',
agents: agentNames.map(name => ({ name })), agents: await listOpenClawAgents(logger),
}; };
const ok = await bridgeClient.pushOpenClawMeta(meta); const ok = await bridgeClient.pushOpenClawMeta(meta);
@@ -175,7 +151,7 @@ export default {
// Fallback: query backend for agent status // Fallback: query backend for agent status
const live = resolveConfig(); const live = resolveConfig();
const agentId = resolveAgentId(); const agentId = process.env.AGENT_ID || 'unknown';
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: {
@@ -195,51 +171,56 @@ export default {
} }
/** /**
* Wake agent via dispatchInboundMessage — same mechanism used by Discord plugin. * Wake/spawn agent with task context for slot execution.
* Direct in-process call, no WebSocket or CLI needed. * This is the callback invoked by CalendarScheduler when a slot is ready.
*/ */
async function wakeAgent(agentId: string): Promise<boolean> { async function wakeAgent(context: AgentWakeContext): Promise<boolean> {
logger.info(`Waking agent ${agentId}: has due slots`); logger.info(`Waking agent for slot: ${context.taskDescription}`);
const sessionKey = `agent:${agentId}:hf-wakeup`;
try { try {
const sdkPath = 'openclaw/plugin-sdk/reply-runtime'; // Method 1: Use OpenClaw spawn API if available (preferred)
const { dispatchInboundMessageWithDispatcher } = await import( if (api.spawn) {
/* webpackIgnore: true */ sdkPath const result = await api.spawn({
); task: context.prompt,
timeoutSeconds: context.slot.estimated_duration * 60, // Convert to seconds
const cfg = api.runtime?.config?.loadConfig?.();
if (!cfg) {
logger.error('Cannot load OpenClaw config for dispatch');
return false;
}
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.`;
const result = await dispatchInboundMessageWithDispatcher({
ctx: {
Body: wakeupMessage,
SessionKey: sessionKey,
From: 'harborforge-calendar',
Provider: 'harborforge',
},
cfg,
dispatcherOptions: {
deliver: async (payload: any) => {
const text = (payload.text || '').trim();
logger.info(`Agent ${agentId} wakeup reply: ${text.slice(0, 100)}`);
},
},
}); });
logger.info(`Agent ${agentId} dispatched: ${result?.status || 'ok'}`); if (result?.sessionId) {
return true; logger.info(`Agent spawned for calendar slot: session=${result.sessionId}`);
} catch (err: any) { // Track session completion
const msg = err?.message || err?.code || String(err); trackSessionCompletion(result.sessionId, context);
const stack = err?.stack?.split('\n').slice(0, 3).join(' | ') || ''; return true;
logger.error(`Failed to dispatch agent for slot: ${msg} ${stack}`); }
}
// Method 2: Send notification/alert to wake agent (fallback)
// 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 live = resolveConfig();
const agentId = process.env.AGENT_ID || 'unknown';
const notifyResponse = await fetch(`${live.backendUrl}/calendar/agent/notify`, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'X-Agent-ID': agentId,
'X-Claw-Identifier': live.identifier || hostname(),
},
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;
} catch (err) {
logger.error('Failed to wake agent:', err);
return false; return false;
} }
} }
@@ -281,69 +262,27 @@ export default {
*/ */
function startCalendarScheduler(): void { function startCalendarScheduler(): void {
const live = resolveConfig(); const live = resolveConfig();
const agentId = process.env.AGENT_ID || 'unknown';
// Create bridge client (claw-instance level, not per-agent) // Create calendar bridge client
const calendarBridge = createCalendarBridgeClient( const calendarBridge = createCalendarBridgeClient(
api, api,
live.backendUrl || 'https://monitor.hangman-lab.top', live.backendUrl || 'https://monitor.hangman-lab.top',
'unused' // agentId no longer needed at bridge level agentId
); );
// Multi-agent sync + check loop // Create and start scheduler
const { MultiAgentScheduleCache } = require('./calendar/schedule-cache') as typeof import('./calendar/schedule-cache'); calendarScheduler = createCalendarScheduler({
const scheduleCache = new MultiAgentScheduleCache(); bridge: calendarBridge,
getAgentStatus,
wakeAgent,
logger,
heartbeatIntervalMs: 60000, // 1 minute
debug: live.logLevel === 'debug',
});
const SYNC_INTERVAL_MS = 300_000; // 5 min calendarScheduler.start();
const CHECK_INTERVAL_MS = 30_000; // 30 sec logger.info('Calendar scheduler started');
// 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)');
} }
/** /**