Compare commits
4 Commits
ec09578de3
...
05bf51930d
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
05bf51930d | ||
|
|
be30b4b3f4 | ||
|
|
7e4750fcc4 | ||
|
|
6d59741086 |
190
REFACTOR_PLAN.md
190
REFACTOR_PLAN.md
@@ -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'
|
||||
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)
|
||||
**Problems:**
|
||||
1. Every heartbeat queries backend for pending slots — no local awareness of full schedule
|
||||
2. Cannot detect slots assigned by other agents between heartbeats
|
||||
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
|
||||
|
||||
### 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
|
||||
// Local schedule cache: { agentId → [slots] }
|
||||
const schedules: Map<string, CalendarSlotResponse[]> = new Map();
|
||||
// Sync: every 5 min — pull full schedule from backend
|
||||
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
|
||||
|
||||
```
|
||||
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)
|
||||
```
|
||||
`runCheck()`:
|
||||
1. `cache.getDueSlots(now)` → find due slots
|
||||
2. Filter out session-deferred slots
|
||||
3. If agent idle → select highest priority → execute
|
||||
|
||||
### Heartbeat (every 60s)
|
||||
### 4. Wakeup via Dirigent (future)
|
||||
|
||||
Simplified to liveness ping only:
|
||||
```
|
||||
POST /monitor/server/heartbeat
|
||||
claw_identifier: xxx
|
||||
→ server returns empty/ack
|
||||
```
|
||||
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
|
||||
- Creating a private channel with only the target agent
|
||||
- Posting the wakeup prompt as a message
|
||||
|
||||
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 }
|
||||
```
|
||||
**For now:** Keep `api.spawn()` as the wakeup method. The Dirigent integration can be added later as it requires cross-plugin coordination.
|
||||
|
||||
## 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)
|
||||
1. Add `ScheduleCache` class (new file: `plugin/calendar/schedule-cache.ts`)
|
||||
2. Add `getDaySchedule()` to `CalendarBridgeClient`
|
||||
3. Refactor `CalendarScheduler`:
|
||||
- Replace single interval with sync + check intervals
|
||||
- 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)
|
||||
- New route: `/calendar/sync`
|
||||
- New service: schedule diff tracking per claw_identifier
|
||||
| File | Changes |
|
||||
|------|---------|
|
||||
| `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
|
||||
- `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
|
||||
## Risk Assessment
|
||||
|
||||
- **Low risk:** ScheduleCache is additive, doesn't break existing behavior
|
||||
- **Medium risk:** Splitting heartbeat changes core scheduling logic
|
||||
- **Mitigation:** Keep `heartbeat()` method intact, use it for liveness reporting alongside new sync
|
||||
|
||||
@@ -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
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@@ -1,97 +1,105 @@
|
||||
/**
|
||||
* 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.
|
||||
* Local cache of today's calendar schedule.
|
||||
* Synced periodically from HF backend, checked locally for due slots.
|
||||
*/
|
||||
|
||||
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;
|
||||
}
|
||||
import type { CalendarSlotResponse } from "./types.js";
|
||||
|
||||
export class MultiAgentScheduleCache {
|
||||
/** { agentId → slots[] } */
|
||||
private schedules: Map<string, CachedSlot[]> = new Map();
|
||||
export class ScheduleCache {
|
||||
private slots: Map<string, CalendarSlotResponse> = new Map();
|
||||
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) {
|
||||
this.schedules.clear();
|
||||
this.slots.clear();
|
||||
}
|
||||
this.cachedDate = date;
|
||||
|
||||
for (const [agentId, slots] of Object.entries(schedules)) {
|
||||
this.schedules.set(agentId, slots);
|
||||
// Merge: update existing slots, add new ones
|
||||
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();
|
||||
}
|
||||
|
||||
/**
|
||||
* Get agents that have due (overdue or current) slots.
|
||||
* Returns [agentId, dueSlots[]] pairs.
|
||||
* Get slots that are due (scheduled_at <= now) and still pending.
|
||||
*/
|
||||
getAgentsWithDueSlots(now: Date): Array<{ agentId: string; slots: CachedSlot[] }> {
|
||||
const results: Array<{ agentId: string; slots: CachedSlot[] }> = [];
|
||||
getDueSlots(now: Date): CalendarSlotResponse[] {
|
||||
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 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) {
|
||||
const scheduledAt = this.parseScheduledTime(slot.scheduled_at);
|
||||
if (scheduledAt && scheduledAt <= now) {
|
||||
results.push(slot);
|
||||
}
|
||||
}
|
||||
// Sort by priority descending
|
||||
due.sort((a, b) => (b.priority ?? 0) - (a.priority ?? 0));
|
||||
results.push({ agentId, slots: due });
|
||||
}
|
||||
}
|
||||
results.sort((a, b) => (b.priority ?? 0) - (a.priority ?? 0));
|
||||
return results;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all agent IDs in the cache.
|
||||
* Update a slot locally (e.g., after status change).
|
||||
*/
|
||||
getAgentIds(): string[] {
|
||||
return Array.from(this.schedules.keys());
|
||||
updateSlot(slotId: string, update: Partial<CalendarSlotResponse>): void {
|
||||
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[] {
|
||||
return this.schedules.get(agentId) ?? [];
|
||||
removeSlot(slotId: string): void {
|
||||
this.slots.delete(slotId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get cache status for debugging.
|
||||
* Get all cached slots.
|
||||
*/
|
||||
getStatus(): { agentCount: number; totalSlots: number; lastSyncAt: string | null; cachedDate: string | null } {
|
||||
let totalSlots = 0;
|
||||
for (const slots of this.schedules.values()) totalSlots += slots.length;
|
||||
getAll(): CalendarSlotResponse[] {
|
||||
return Array.from(this.slots.values());
|
||||
}
|
||||
|
||||
/**
|
||||
* Get cache metadata.
|
||||
*/
|
||||
getStatus(): { slotCount: number; lastSyncAt: string | null; cachedDate: string | null } {
|
||||
return {
|
||||
agentCount: this.schedules.size,
|
||||
totalSlots,
|
||||
slotCount: this.slots.size,
|
||||
lastSyncAt: this.lastSyncAt?.toISOString() ?? null,
|
||||
cachedDate: this.cachedDate,
|
||||
};
|
||||
}
|
||||
|
||||
private getSlotId(slot: CalendarSlotResponse): string {
|
||||
return slot.virtual_id ?? String(slot.id);
|
||||
}
|
||||
|
||||
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)) {
|
||||
// Time-only: combine with cached date
|
||||
if (!this.cachedDate) return null;
|
||||
return new Date(`${this.cachedDate}T${scheduledAt}Z`);
|
||||
}
|
||||
|
||||
@@ -1,3 +1,8 @@
|
||||
import { execFile } from 'child_process';
|
||||
import { promisify } from 'util';
|
||||
|
||||
const execFileAsync = promisify(execFile);
|
||||
|
||||
export interface OpenClawAgentInfo {
|
||||
name: string;
|
||||
isDefault?: boolean;
|
||||
@@ -9,38 +14,70 @@ export interface OpenClawAgentInfo {
|
||||
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 [];
|
||||
}
|
||||
}
|
||||
|
||||
export function parseOpenClawAgents(text: string): OpenClawAgentInfo[] {
|
||||
const lines = text.split(/\r?\n/);
|
||||
const out: OpenClawAgentInfo[] = [];
|
||||
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) {
|
||||
const line = raw.trimEnd();
|
||||
if (!line.trim() || line.startsWith("Agents:") || line.startsWith("Routing rules map") || line.startsWith("Channel status reflects")) continue;
|
||||
if (line.startsWith("- ")) {
|
||||
if (!line.trim() || line.startsWith('Agents:') || line.startsWith('Routing rules map') || line.startsWith('Channel status reflects')) continue;
|
||||
if (line.startsWith('- ')) {
|
||||
push();
|
||||
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;
|
||||
}
|
||||
if (!current) continue;
|
||||
const trimmed = line.trim();
|
||||
const idx = trimmed.indexOf(":");
|
||||
const idx = trimmed.indexOf(':');
|
||||
if (idx === -1) continue;
|
||||
const key = trimmed.slice(0, idx).trim();
|
||||
const value = trimmed.slice(idx + 1).trim();
|
||||
switch (key) {
|
||||
case "Identity": current.identity = value; break;
|
||||
case "Workspace": current.workspace = value; 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;
|
||||
case 'Identity':
|
||||
current.identity = value;
|
||||
break;
|
||||
case 'Workspace':
|
||||
current.workspace = value;
|
||||
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();
|
||||
|
||||
187
plugin/index.ts
187
plugin/index.ts
@@ -14,7 +14,7 @@
|
||||
import { hostname, freemem, totalmem, uptime, loadavg, platform } from 'os';
|
||||
import { getPluginConfig } from './core/config';
|
||||
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 { registerGatewayStopHook } from './hooks/gateway-stop';
|
||||
import {
|
||||
@@ -32,12 +32,6 @@ interface PluginAPI {
|
||||
warn: (...args: any[]) => void;
|
||||
};
|
||||
version?: string;
|
||||
runtime?: {
|
||||
version?: string;
|
||||
config?: {
|
||||
loadConfig?: () => any;
|
||||
};
|
||||
};
|
||||
config?: Record<string, unknown>;
|
||||
pluginConfig?: Record<string, unknown>;
|
||||
on: (event: string, handler: () => void) => void;
|
||||
@@ -68,13 +62,6 @@ export default {
|
||||
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.
|
||||
*/
|
||||
@@ -109,7 +96,7 @@ export default {
|
||||
avg15: load[2],
|
||||
},
|
||||
openclaw: {
|
||||
version: api.runtime?.version || api.version || 'unknown',
|
||||
version: api.version || 'unknown',
|
||||
pluginVersion: '0.3.1', // Bumped for PLG-CAL-004
|
||||
},
|
||||
timestamp: new Date().toISOString(),
|
||||
@@ -131,21 +118,10 @@ export default {
|
||||
const bridgeClient = getBridgeClient();
|
||||
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 = {
|
||||
version: api.runtime?.version || api.version || 'unknown',
|
||||
version: api.version || 'unknown',
|
||||
plugin_version: '0.3.1',
|
||||
agents: agentNames.map(name => ({ name })),
|
||||
agents: await listOpenClawAgents(logger),
|
||||
};
|
||||
|
||||
const ok = await bridgeClient.pushOpenClawMeta(meta);
|
||||
@@ -175,7 +151,7 @@ export default {
|
||||
|
||||
// Fallback: query backend for agent status
|
||||
const live = resolveConfig();
|
||||
const agentId = resolveAgentId();
|
||||
const agentId = process.env.AGENT_ID || 'unknown';
|
||||
try {
|
||||
const response = await fetch(`${live.backendUrl}/calendar/agent/status?agent_id=${agentId}`, {
|
||||
headers: {
|
||||
@@ -195,51 +171,56 @@ export default {
|
||||
}
|
||||
|
||||
/**
|
||||
* Wake agent via dispatchInboundMessage — same mechanism used by Discord plugin.
|
||||
* Direct in-process call, no WebSocket or CLI needed.
|
||||
* Wake/spawn agent with task context for slot execution.
|
||||
* This is the callback invoked by CalendarScheduler when a slot is ready.
|
||||
*/
|
||||
async function wakeAgent(agentId: string): Promise<boolean> {
|
||||
logger.info(`Waking agent ${agentId}: has due slots`);
|
||||
|
||||
const sessionKey = `agent:${agentId}:hf-wakeup`;
|
||||
async function wakeAgent(context: AgentWakeContext): Promise<boolean> {
|
||||
logger.info(`Waking agent for slot: ${context.taskDescription}`);
|
||||
|
||||
try {
|
||||
const sdkPath = 'openclaw/plugin-sdk/reply-runtime';
|
||||
const { dispatchInboundMessageWithDispatcher } = await import(
|
||||
/* webpackIgnore: true */ sdkPath
|
||||
);
|
||||
|
||||
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)}`);
|
||||
},
|
||||
},
|
||||
// Method 1: Use OpenClaw spawn API if available (preferred)
|
||||
if (api.spawn) {
|
||||
const result = await api.spawn({
|
||||
task: context.prompt,
|
||||
timeoutSeconds: context.slot.estimated_duration * 60, // Convert to seconds
|
||||
});
|
||||
|
||||
logger.info(`Agent ${agentId} dispatched: ${result?.status || 'ok'}`);
|
||||
return true;
|
||||
if (result?.sessionId) {
|
||||
logger.info(`Agent spawned for calendar slot: session=${result.sessionId}`);
|
||||
|
||||
} catch (err: any) {
|
||||
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}`);
|
||||
// Track session completion
|
||||
trackSessionCompletion(result.sessionId, context);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
// 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;
|
||||
}
|
||||
}
|
||||
@@ -281,69 +262,27 @@ export default {
|
||||
*/
|
||||
function startCalendarScheduler(): void {
|
||||
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(
|
||||
api,
|
||||
live.backendUrl || 'https://monitor.hangman-lab.top',
|
||||
'unused' // agentId no longer needed at bridge level
|
||||
agentId
|
||||
);
|
||||
|
||||
// Multi-agent sync + check loop
|
||||
const { MultiAgentScheduleCache } = require('./calendar/schedule-cache') as typeof import('./calendar/schedule-cache');
|
||||
const scheduleCache = new MultiAgentScheduleCache();
|
||||
// Create and start scheduler
|
||||
calendarScheduler = createCalendarScheduler({
|
||||
bridge: calendarBridge,
|
||||
getAgentStatus,
|
||||
wakeAgent,
|
||||
logger,
|
||||
heartbeatIntervalMs: 60000, // 1 minute
|
||||
debug: live.logLevel === 'debug',
|
||||
});
|
||||
|
||||
const SYNC_INTERVAL_MS = 300_000; // 5 min
|
||||
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)');
|
||||
calendarScheduler.start();
|
||||
logger.info('Calendar scheduler started');
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user