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'
|
**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
|
||||||
|
|||||||
@@ -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
|
||||||
// -------------------------------------------------------------------------
|
// -------------------------------------------------------------------------
|
||||||
|
|||||||
@@ -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`);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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();
|
||||||
|
|||||||
187
plugin/index.ts
187
plugin/index.ts
@@ -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)');
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
Reference in New Issue
Block a user