zhi/HarborForge.OpenclawPlugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

PLG-CAL-004: Implement ScheduledGatewayRestart handling in plugin

Browse Source 
- Add state persistence (persistState/restoreState) for recovery after restart
- Add handleScheduledGatewayRestart method that:
  - Persists current scheduler state to disk
  - Sends final heartbeat to backend before shutdown
  - Stops the calendar scheduler (pauses scheduled tasks)
- Add isRestartPending flag to prevent new slot processing during restart
- Add isScheduledGatewayRestart helper to detect restart events
- Update scheduler to detect and handle ScheduledGatewayRestart events
- Add new tools: harborforge_restart_status, harborforge_calendar_pause/resume
- Export isRestartPending and getStateFilePath methods
- Bump plugin version to 0.3.1
This commit is contained in:
zhi
2026-04-01 09:41:02 +00:00
parent 24c4a7ad14
commit 3b0ea0ad12
6 changed files with 578 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
plugin/calendar/scheduler.d.ts vendored
View File

@@ -2,6 +2,7 @@
* HarborForge Calendar Scheduler
*
* PLG-CAL-002: Plugin-side handling for pending slot execution.
* PLG-CAL-004: ScheduledGatewayRestart event handling with state persistence.
*
* Responsibilities:
* - Run calendar heartbeat every minute
@@ -9,6 +10,8 @@
* - Wake agent with task context
* - Handle slot status transitions (attended, ongoing, deferred)
* - Manage agent status transitions (idle → busy/on_call)
* - Persist state on ScheduledGatewayRestart and restore on startup
* - Send final heartbeat before graceful shutdown
*
* Design reference: NEXT_WAVE_DEV_DIRECTION.md §6 (Agent wakeup mechanism)
*/
@@ -32,6 +35,8 @@ export interface CalendarSchedulerConfig {
heartbeatIntervalMs?: number;
/** Enable verbose debug logging */
debug?: boolean;
/** Directory for state persistence (default: plugin data dir) */
stateDir?: string;
}
/**
* Context passed to agent when waking for slot execution.
@@ -63,6 +68,8 @@ interface SchedulerState {
deferredSlotIds: Set<string>;
/** Whether agent is currently processing a slot */
isProcessing: boolean;
/** Whether a gateway restart is scheduled/pending */
isRestartPending: boolean;
}
/**
* CalendarScheduler manages the periodic heartbeat and slot execution lifecycle.
@@ -70,7 +77,33 @@ interface SchedulerState {
export declare class CalendarScheduler {
private config;
private state;
private stateFilePath;
constructor(config: CalendarSchedulerConfig);
/**
* Get default state directory (plugin data directory or temp fallback).
*/
private getDefaultStateDir;
/**
* Persist current state to disk for recovery after restart.
*/
private persistState;
/**
* Restore state from disk if available.
*/
private restoreState;
/**
* Clear persisted state file after successful restore.
*/
private clearPersistedState;
/**
* Send a final heartbeat to the backend before shutdown.
*/
private sendFinalHeartbeat;
/**
* Handle ScheduledGatewayRestart event.
* PLG-CAL-004: Persist state, send final heartbeat, pause scheduled tasks.
*/
private handleScheduledGatewayRestart;
/**
* Start the calendar scheduler.
* Begins periodic heartbeat to check for pending slots.
@@ -96,6 +129,10 @@ export declare class CalendarScheduler {
* Select highest priority slot and wake agent.
*/
private handleIdleAgent;
/**
* Check if a slot is a ScheduledGatewayRestart system event.
*/
private isScheduledGatewayRestart;
/**
* Execute a slot by waking the agent.
*/
@@ -181,6 +218,14 @@ export declare class CalendarScheduler {
* Get the current slot being executed (if any).
*/
getCurrentSlot(): CalendarSlotResponse | null;
/**
* Check if a gateway restart is pending.
*/
isRestartPending(): boolean;
/**
* Get the path to the state file.
*/
getStateFilePath(): string;
}
/**
* Factory function to create a CalendarScheduler from plugin context.

2
plugin/calendar/scheduler.d.ts.map
View File

@@ -1 +1 @@
{"version":3,"file":"scheduler.d.ts","sourceRoot":"","sources":["scheduler.ts"],"names":[],"mappings":"AAAA;;;;;;;;;;;;;GAaG;AAEH,OAAO,EACL,oBAAoB,EACrB,MAAM,mBAAmB,CAAC;AAC3B,OAAO,EACL,oBAAoB,EAEpB,gBAAgB,EAIjB,MAAM,SAAS,CAAC;AAEjB,MAAM,WAAW,uBAAuB;IACtC,uDAAuD;IACvD,MAAM,EAAE,oBAAoB,CAAC;IAC7B,wDAAwD;IACxD,cAAc,EAAE,MAAM,OAAO,CAAC,gBAAgB,GAAG,IAAI,CAAC,CAAC;IACvD,qDAAqD;IACrD,SAAS,EAAE,CAAC,OAAO,EAAE,gBAAgB,KAAK,OAAO,CAAC,OAAO,CAAC,CAAC;IAC3D,sBAAsB;IACtB,MAAM,EAAE;QACN,IAAI,EAAE,CAAC,GAAG,IAAI,EAAE,GAAG,EAAE,KAAK,IAAI,CAAC;QAC/B,KAAK,EAAE,CAAC,GAAG,IAAI,EAAE,GAAG,EAAE,KAAK,IAAI,CAAC;QAChC,KAAK,EAAE,CAAC,GAAG,IAAI,EAAE,GAAG,EAAE,KAAK,IAAI,CAAC;QAChC,IAAI,EAAE,CAAC,GAAG,IAAI,EAAE,GAAG,EAAE,KAAK,IAAI,CAAC;KAChC,CAAC;IACF,0DAA0D;IAC1D,mBAAmB,CAAC,EAAE,MAAM,CAAC;IAC7B,mCAAmC;IACnC,KAAK,CAAC,EAAE,OAAO,CAAC;CACjB;AAED;;;GAGG;AACH,MAAM,WAAW,gBAAgB;IAC/B,0BAA0B;IAC1B,IAAI,EAAE,oBAAoB,CAAC;IAC3B,sCAAsC;IACtC,eAAe,EAAE,MAAM,CAAC;IACxB,wCAAwC;IACxC,MAAM,EAAE,MAAM,CAAC;IACf,6DAA6D;IAC7D,SAAS,EAAE,OAAO,CAAC;CACpB;AAED;;GAEG;AACH,UAAU,cAAc;IACtB,6CAA6C;IAC7C,SAAS,EAAE,OAAO,CAAC;IACnB,8CAA8C;IAC9C,WAAW,EAAE,oBAAoB,GAAG,IAAI,CAAC;IACzC,+BAA+B;IAC/B,eAAe,EAAE,IAAI,GAAG,IAAI,CAAC;IAC7B,kCAAkC;IAClC,cAAc,EAAE,UAAU,CAAC,OAAO,WAAW,CAAC,GAAG,IAAI,CAAC;IACtD,iEAAiE;IACjE,eAAe,EAAE,GAAG,CAAC,MAAM,CAAC,CAAC;IAC7B,mDAAmD;IACnD,YAAY,EAAE,OAAO,CAAC;CACvB;AAED;;GAEG;AACH,qBAAa,iBAAiB;IAC5B,OAAO,CAAC,MAAM,CAAoC;IAClD,OAAO,CAAC,KAAK,CAAiB;gBAElB,MAAM,EAAE,uBAAuB;IAgB3C;;;OAGG;IACH,KAAK,IAAI,IAAI;IAmBb;;;OAGG;IACH,IAAI,IAAI,IAAI;IAWZ;;;OAGG;IACG,YAAY,IAAI,OAAO,CAAC,IAAI,CAAC;IAyCnC;;;OAGG;YACW,kBAAkB;IA0BhC;;;OAGG;YACW,eAAe;IAiC7B;;OAEG;YACW,WAAW;IAqEzB;;OAEG;IACH,OAAO,CAAC,gBAAgB;IAgCxB;;OAEG;IACH,OAAO,CAAC,cAAc;IAoBtB;;OAEG;IACH,OAAO,CAAC,iBAAiB;IAwCzB;;OAEG;IACH,OAAO,CAAC,wBAAwB;IAShC;;OAEG;IACH,OAAO,CAAC,kBAAkB;IAU1B;;OAEG;YACW,SAAS;IAiBvB;;OAEG;YACW,UAAU;IAiBxB;;;OAGG;IACG,mBAAmB,CAAC,qBAAqB,EAAE,MAAM,GAAG,OAAO,CAAC,IAAI,CAAC;IAmCvE;;;OAGG;IACG,gBAAgB,CAAC,MAAM,CAAC,EAAE,MAAM,GAAG,OAAO,CAAC,IAAI,CAAC;IAkCtD;;;OAGG;IACG,gBAAgB,IAAI,OAAO,CAAC,IAAI,CAAC;IAyBvC;;OAEG;IACG,iBAAiB,IAAI,OAAO,CAAC,IAAI,CAAC;IAyBxC;;;;OAIG;YACW,aAAa;IAc3B;;OAEG;IACH,OAAO,CAAC,SAAS;IAIjB;;OAEG;IACH,OAAO,CAAC,UAAU;IAIlB;;OAEG;IACH,OAAO,CAAC,QAAQ;IAMhB;;OAEG;IACH,QAAQ,IAAI,QAAQ,CAAC,cAAc,CAAC;IAIpC;;OAEG;IACH,SAAS,IAAI,OAAO;IAIpB;;OAEG;IACH,YAAY,IAAI,OAAO;IAIvB;;OAEG;IACH,cAAc,IAAI,oBAAoB,GAAG,IAAI;CAG9C;AAED;;GAEG;AACH,wBAAgB,uBAAuB,CACrC,MAAM,EAAE,uBAAuB,GAC9B,iBAAiB,CAEnB"}
{"version":3,"file":"scheduler.d.ts","sourceRoot":"","sources":["scheduler.ts"],"names":[],"mappings":"AAAA;;;;;;;;;;;;;;;;GAgBG;AAIH,OAAO,EAAE,oBAAoB,EAAE,MAAM,mBAAmB,CAAC;AACzD,OAAO,EACL,oBAAoB,EAEpB,gBAAgB,EAIjB,MAAM,SAAS,CAAC;AAEjB,MAAM,WAAW,uBAAuB;IACtC,uDAAuD;IACvD,MAAM,EAAE,oBAAoB,CAAC;IAC7B,wDAAwD;IACxD,cAAc,EAAE,MAAM,OAAO,CAAC,gBAAgB,GAAG,IAAI,CAAC,CAAC;IACvD,qDAAqD;IACrD,SAAS,EAAE,CAAC,OAAO,EAAE,gBAAgB,KAAK,OAAO,CAAC,OAAO,CAAC,CAAC;IAC3D,sBAAsB;IACtB,MAAM,EAAE;QACN,IAAI,EAAE,CAAC,GAAG,IAAI,EAAE,GAAG,EAAE,KAAK,IAAI,CAAC;QAC/B,KAAK,EAAE,CAAC,GAAG,IAAI,EAAE,GAAG,EAAE,KAAK,IAAI,CAAC;QAChC,KAAK,EAAE,CAAC,GAAG,IAAI,EAAE,GAAG,EAAE,KAAK,IAAI,CAAC;QAChC,IAAI,EAAE,CAAC,GAAG,IAAI,EAAE,GAAG,EAAE,KAAK,IAAI,CAAC;KAChC,CAAC;IACF,0DAA0D;IAC1D,mBAAmB,CAAC,EAAE,MAAM,CAAC;IAC7B,mCAAmC;IACnC,KAAK,CAAC,EAAE,OAAO,CAAC;IAChB,iEAAiE;IACjE,QAAQ,CAAC,EAAE,MAAM,CAAC;CACnB;AAED;;;GAGG;AACH,MAAM,WAAW,gBAAgB;IAC/B,0BAA0B;IAC1B,IAAI,EAAE,oBAAoB,CAAC;IAC3B,sCAAsC;IACtC,eAAe,EAAE,MAAM,CAAC;IACxB,wCAAwC;IACxC,MAAM,EAAE,MAAM,CAAC;IACf,6DAA6D;IAC7D,SAAS,EAAE,OAAO,CAAC;CACpB;AAsBD;;GAEG;AACH,UAAU,cAAc;IACtB,6CAA6C;IAC7C,SAAS,EAAE,OAAO,CAAC;IACnB,8CAA8C;IAC9C,WAAW,EAAE,oBAAoB,GAAG,IAAI,CAAC;IACzC,+BAA+B;IAC/B,eAAe,EAAE,IAAI,GAAG,IAAI,CAAC;IAC7B,kCAAkC;IAClC,cAAc,EAAE,UAAU,CAAC,OAAO,WAAW,CAAC,GAAG,IAAI,CAAC;IACtD,iEAAiE;IACjE,eAAe,EAAE,GAAG,CAAC,MAAM,CAAC,CAAC;IAC7B,mDAAmD;IACnD,YAAY,EAAE,OAAO,CAAC;IACtB,qDAAqD;IACrD,gBAAgB,EAAE,OAAO,CAAC;CAC3B;AAOD;;GAEG;AACH,qBAAa,iBAAiB;IAC5B,OAAO,CAAC,MAAM,CAAoC;IAClD,OAAO,CAAC,KAAK,CAAiB;IAC9B,OAAO,CAAC,aAAa,CAAS;gBAElB,MAAM,EAAE,uBAAuB;IAwB3C;;OAEG;IACH,OAAO,CAAC,kBAAkB;IA8B1B;;OAEG;IACH,OAAO,CAAC,YAAY;IAmBpB;;OAEG;IACH,OAAO,CAAC,YAAY;IAuCpB;;OAEG;IACH,OAAO,CAAC,mBAAmB;IAY3B;;OAEG;YACW,kBAAkB;IAahC;;;OAGG;YACW,6BAA6B;IAuC3C;;;OAGG;IACH,KAAK,IAAI,IAAI;IAoBb;;;OAGG;IACH,IAAI,IAAI,IAAI;IAWZ;;;OAGG;IACG,YAAY,IAAI,OAAO,CAAC,IAAI,CAAC;IAgDnC;;;OAGG;YACW,kBAAkB;IA0BhC;;;OAGG;YACW,eAAe;IAuC7B;;OAEG;IACH,OAAO,CAAC,yBAAyB;IAQjC;;OAEG;YACW,WAAW;IAoEzB;;OAEG;IACH,OAAO,CAAC,gBAAgB;IAgCxB;;OAEG;IACH,OAAO,CAAC,cAAc;IAuBtB;;OAEG;IACH,OAAO,CAAC,iBAAiB;IAwCzB;;OAEG;IACH,OAAO,CAAC,wBAAwB;IAShC;;OAEG;IACH,OAAO,CAAC,kBAAkB;IAU1B;;OAEG;YACW,SAAS;IAiBvB;;OAEG;YACW,UAAU;IAiBxB;;;OAGG;IACG,mBAAmB,CAAC,qBAAqB,EAAE,MAAM,GAAG,OAAO,CAAC,IAAI,CAAC;IAkCvE;;;OAGG;IACG,gBAAgB,CAAC,MAAM,CAAC,EAAE,MAAM,GAAG,OAAO,CAAC,IAAI,CAAC;IAiCtD;;;OAGG;IACG,gBAAgB,IAAI,OAAO,CAAC,IAAI,CAAC;IAwBvC;;OAEG;IACG,iBAAiB,IAAI,OAAO,CAAC,IAAI,CAAC;IAwBxC;;;;OAIG;YACW,aAAa;IAc3B;;OAEG;IACH,OAAO,CAAC,SAAS;IAIjB;;OAEG;IACH,OAAO,CAAC,UAAU;IAIlB;;OAEG;IACH,OAAO,CAAC,QAAQ;IAMhB;;OAEG;IACH,QAAQ,IAAI,QAAQ,CAAC,cAAc,CAAC;IAIpC;;OAEG;IACH,SAAS,IAAI,OAAO;IAIpB;;OAEG;IACH,YAAY,IAAI,OAAO;IAIvB;;OAEG;IACH,cAAc,IAAI,oBAAoB,GAAG,IAAI;IAI7C;;OAEG;IACH,gBAAgB,IAAI,OAAO;IAI3B;;OAEG;IACH,gBAAgB,IAAI,MAAM;CAG3B;AAED;;GAEG;AACH,wBAAgB,uBAAuB,CACrC,MAAM,EAAE,uBAAuB,GAC9B,iBAAiB,CAEnB"}

202
plugin/calendar/scheduler.js
View File

@@ -3,6 +3,7 @@
* HarborForge Calendar Scheduler
*
* PLG-CAL-002: Plugin-side handling for pending slot execution.
* PLG-CAL-004: ScheduledGatewayRestart event handling with state persistence.
*
* Responsibilities:
* - Run calendar heartbeat every minute
@@ -10,25 +11,36 @@
* - Wake agent with task context
* - Handle slot status transitions (attended, ongoing, deferred)
* - Manage agent status transitions (idle → busy/on_call)
* - Persist state on ScheduledGatewayRestart and restore on startup
* - Send final heartbeat before graceful shutdown
*
* Design reference: NEXT_WAVE_DEV_DIRECTION.md §6 (Agent wakeup mechanism)
*/
Object.defineProperty(exports, "__esModule", { value: true });
exports.CalendarScheduler = void 0;
exports.createCalendarScheduler = createCalendarScheduler;
const fs_1 = require("fs");
const path_1 = require("path");
const types_1 = require("./types");
/** State file name */
const STATE_FILENAME = 'calendar-scheduler-state.json';
/** State file version for migration compatibility */
const STATE_VERSION = 1;
/**
* CalendarScheduler manages the periodic heartbeat and slot execution lifecycle.
*/
class CalendarScheduler {
config;
state;
stateFilePath;
constructor(config) {
this.config = {
heartbeatIntervalMs: 60000, // 1 minute default
debug: false,
stateDir: this.getDefaultStateDir(),
...config,
};
this.stateFilePath = (0, path_1.join)(this.config.stateDir, STATE_FILENAME);
this.state = {
isRunning: false,
currentSlot: null,
@@ -36,7 +48,162 @@ class CalendarScheduler {
intervalHandle: null,
deferredSlotIds: new Set(),
isProcessing: false,
isRestartPending: false,
};
// Attempt to restore state from previous persistence
this.restoreState();
}
/**
* Get default state directory (plugin data directory or temp fallback).
*/
getDefaultStateDir() {
// Try to use the plugin's directory or a standard data location
const candidates = [
process.env.OPENCLAW_PLUGIN_DATA_DIR,
process.env.HARBORFORGE_PLUGIN_DIR,
(0, path_1.join)(process.cwd(), '.harborforge'),
(0, path_1.join)(process.cwd(), 'data'),
'/tmp/harborforge',
];
for (const dir of candidates) {
if (dir) {
try {
if (!(0, fs_1.existsSync)(dir)) {
(0, fs_1.mkdirSync)(dir, { recursive: true });
}
// Test write access
const testFile = (0, path_1.join)(dir, '.write-test');
(0, fs_1.writeFileSync)(testFile, '', { flag: 'w' });
return dir;
}
catch {
continue;
}
}
}
// Fallback to current working directory
return process.cwd();
}
/**
* Persist current state to disk for recovery after restart.
*/
persistState(reason) {
try {
const persistedState = {
version: STATE_VERSION,
persistedAt: new Date().toISOString(),
reason,
currentSlot: this.state.currentSlot,
deferredSlotIds: Array.from(this.state.deferredSlotIds),
isProcessing: this.state.isProcessing,
agentStatus: null, // Will be determined at restore time
};
(0, fs_1.writeFileSync)(this.stateFilePath, JSON.stringify(persistedState, null, 2));
this.config.logger.info(`[PLG-CAL-004] State persisted to ${this.stateFilePath} (reason: ${reason})`);
}
catch (err) {
this.config.logger.error('[PLG-CAL-004] Failed to persist state:', err);
}
}
/**
* Restore state from disk if available.
*/
restoreState() {
try {
if (!(0, fs_1.existsSync)(this.stateFilePath)) {
return;
}
const data = (0, fs_1.readFileSync)(this.stateFilePath, 'utf-8');
const persisted = JSON.parse(data);
// Validate version
if (persisted.version !== STATE_VERSION) {
this.config.logger.warn(`[PLG-CAL-004] State version mismatch: ${persisted.version} vs ${STATE_VERSION}`);
this.clearPersistedState();
return;
}
// Restore deferred slot IDs
if (persisted.deferredSlotIds && persisted.deferredSlotIds.length > 0) {
this.state.deferredSlotIds = new Set(persisted.deferredSlotIds);
this.config.logger.info(`[PLG-CAL-004] Restored ${persisted.deferredSlotIds.length} deferred slot(s)`);
}
// If there was a slot in progress, mark it for replanning
if (persisted.isProcessing && persisted.currentSlot) {
this.config.logger.warn(`[PLG-CAL-004] Previous session had in-progress slot: ${this.getSlotId(persisted.currentSlot)}`);
// The slot will be picked up by the next heartbeat and can be resumed or deferred
}
this.config.logger.info(`[PLG-CAL-004] State restored from ${persisted.persistedAt} (reason: ${persisted.reason})`);
// Clear the persisted state after successful restore
this.clearPersistedState();
}
catch (err) {
this.config.logger.error('[PLG-CAL-004] Failed to restore state:', err);
}
}
/**
* Clear persisted state file after successful restore.
*/
clearPersistedState() {
try {
if ((0, fs_1.existsSync)(this.stateFilePath)) {
// In a real implementation, we might want to archive instead of delete
// For now, we'll just clear the content to mark as processed
(0, fs_1.writeFileSync)(this.stateFilePath, JSON.stringify({ restored: true, at: new Date().toISOString() }));
}
}
catch (err) {
this.config.logger.error('[PLG-CAL-004] Failed to clear persisted state:', err);
}
}
/**
* Send a final heartbeat to the backend before shutdown.
*/
async sendFinalHeartbeat(reason) {
try {
this.config.logger.info(`[PLG-CAL-004] Sending final heartbeat (reason: ${reason})`);
// Send agent status update indicating we're going offline
await this.config.bridge.reportAgentStatus({ status: 'offline' });
this.config.logger.info('[PLG-CAL-004] Final heartbeat sent successfully');
}
catch (err) {
this.config.logger.error('[PLG-CAL-004] Failed to send final heartbeat:', err);
}
}
/**
* Handle ScheduledGatewayRestart event.
* PLG-CAL-004: Persist state, send final heartbeat, pause scheduled tasks.
*/
async handleScheduledGatewayRestart(slot) {
this.config.logger.info('[PLG-CAL-004] Handling ScheduledGatewayRestart event');
// 1. Mark restart as pending to prevent new slot processing
this.state.isRestartPending = true;
// 2. Persist current state
this.persistState('ScheduledGatewayRestart');
// 3. If there's a current slot, pause it gracefully
if (this.state.isProcessing && this.state.currentSlot) {
this.config.logger.info('[PLG-CAL-004] Pausing current slot before restart');
await this.pauseCurrentSlot();
}
// 4. Send final heartbeat
await this.sendFinalHeartbeat('ScheduledGatewayRestart');
// 5. Stop the scheduler (pause scheduled tasks)
this.config.logger.info('[PLG-CAL-004] Stopping scheduler due to gateway restart');
this.stop();
// 6. Mark the slot as finished (since we've handled the restart)
const update = {
status: types_1.SlotStatus.FINISHED,
actual_duration: 0, // Restart preparation doesn't take time
};
try {
if (slot.id) {
await this.config.bridge.updateSlot(slot.id, update);
}
else if (slot.virtual_id) {
await this.config.bridge.updateVirtualSlot(slot.virtual_id, update);
}
}
catch (err) {
this.config.logger.error('[PLG-CAL-004] Failed to mark restart slot as finished:', err);
}
}
/**
* Start the calendar scheduler.
@@ -48,6 +215,7 @@ class CalendarScheduler {
return;
}
this.state.isRunning = true;
this.state.isRestartPending = false;
this.config.logger.info('Calendar scheduler started');
// Run initial heartbeat immediately
this.runHeartbeat();
@@ -74,6 +242,11 @@ class CalendarScheduler {
if (!this.state.isRunning) {
return;
}
// Skip heartbeat if restart is pending
if (this.state.isRestartPending) {
this.logDebug('Heartbeat skipped: gateway restart pending');
return;
}
this.state.lastHeartbeatAt = new Date();
try {
// Fetch pending slots from backend
@@ -130,7 +303,7 @@ class CalendarScheduler {
return;
}
// Filter out already deferred slots in this session
const eligibleSlots = slots.filter(s => !this.state.deferredSlotIds.has(this.getSlotId(s)));
const eligibleSlots = slots.filter((s) => !this.state.deferredSlotIds.has(this.getSlotId(s)));
if (eligibleSlots.length === 0) {
this.logDebug('All pending slots have been deferred this session');
return;
@@ -144,9 +317,24 @@ class CalendarScheduler {
await this.deferSlot(slot);
this.state.deferredSlotIds.add(this.getSlotId(slot));
}
// Check if this is a ScheduledGatewayRestart event
if (this.isScheduledGatewayRestart(selectedSlot)) {
await this.handleScheduledGatewayRestart(selectedSlot);
return;
}
// Wake agent to execute selected slot
await this.executeSlot(selectedSlot);
}
/**
* Check if a slot is a ScheduledGatewayRestart system event.
*/
isScheduledGatewayRestart(slot) {
if (slot.event_type !== 'system_event' || !slot.event_data) {
return false;
}
const sysData = slot.event_data;
return sysData.event === 'ScheduledGatewayRestart';
}
/**
* Execute a slot by waking the agent.
*/
@@ -544,6 +732,18 @@ Please use this time for the scheduled activity.`;
getCurrentSlot() {
return this.state.currentSlot;
}
/**
* Check if a gateway restart is pending.
*/
isRestartPending() {
return this.state.isRestartPending;
}
/**
* Get the path to the state file.
*/
getStateFilePath() {
return this.stateFilePath;
}
}
exports.CalendarScheduler = CalendarScheduler;
/**

2
plugin/calendar/scheduler.js.map
View File

File diff suppressed because one or more lines are too long

275
plugin/calendar/scheduler.ts
View File

@@ -2,6 +2,7 @@
* HarborForge Calendar Scheduler
*
* PLG-CAL-002: Plugin-side handling for pending slot execution.
* PLG-CAL-004: ScheduledGatewayRestart event handling with state persistence.
*
* Responsibilities:
* - Run calendar heartbeat every minute
@@ -9,13 +10,15 @@
* - Wake agent with task context
* - Handle slot status transitions (attended, ongoing, deferred)
* - Manage agent status transitions (idle → busy/on_call)
* - Persist state on ScheduledGatewayRestart and restore on startup
* - Send final heartbeat before graceful shutdown
*
* Design reference: NEXT_WAVE_DEV_DIRECTION.md §6 (Agent wakeup mechanism)
*/
import {
CalendarBridgeClient,
} from './calendar-bridge';
import { writeFileSync, readFileSync, existsSync, mkdirSync } from 'fs';
import { join, dirname } from 'path';
import { CalendarBridgeClient } from './calendar-bridge';
import {
CalendarSlotResponse,
SlotStatus,
@@ -43,6 +46,8 @@ export interface CalendarSchedulerConfig {
heartbeatIntervalMs?: number;
/** Enable verbose debug logging */
debug?: boolean;
/** Directory for state persistence (default: plugin data dir) */
stateDir?: string;
}
/**
@@ -60,6 +65,26 @@ export interface AgentWakeContext {
isVirtual: boolean;
}
/**
* Persisted state structure for recovery after restart.
*/
interface PersistedState {
/** Version for migration compatibility */
version: number;
/** When the state was persisted */
persistedAt: string;
/** Reason for persistence (e.g., 'ScheduledGatewayRestart') */
reason: string;
/** The slot that was being executed when persisted */
currentSlot: CalendarSlotResponse | null;
/** Deferred slot IDs at persistence time */
deferredSlotIds: string[];
/** Whether a slot was in progress */
isProcessing: boolean;
/** Agent status at persistence time */
agentStatus: AgentStatusValue | null;
}
/**
* Current execution state tracked by the scheduler.
*/
@@ -76,21 +101,33 @@ interface SchedulerState {
deferredSlotIds: Set<string>;
/** Whether agent is currently processing a slot */
isProcessing: boolean;
/** Whether a gateway restart is scheduled/pending */
isRestartPending: boolean;
}
/** State file name */
const STATE_FILENAME = 'calendar-scheduler-state.json';
/** State file version for migration compatibility */
const STATE_VERSION = 1;
/**
* CalendarScheduler manages the periodic heartbeat and slot execution lifecycle.
*/
export class CalendarScheduler {
private config: Required<CalendarSchedulerConfig>;
private state: SchedulerState;
private stateFilePath: string;
constructor(config: CalendarSchedulerConfig) {
this.config = {
heartbeatIntervalMs: 60000, // 1 minute default
debug: false,
stateDir: this.getDefaultStateDir(),
...config,
};
this.stateFilePath = join(this.config.stateDir, STATE_FILENAME);
this.state = {
isRunning: false,
currentSlot: null,
@@ -98,7 +135,182 @@ export class CalendarScheduler {
intervalHandle: null,
deferredSlotIds: new Set(),
isProcessing: false,
isRestartPending: false,
};
// Attempt to restore state from previous persistence
this.restoreState();
}
/**
* Get default state directory (plugin data directory or temp fallback).
*/
private getDefaultStateDir(): string {
// Try to use the plugin's directory or a standard data location
const candidates = [
process.env.OPENCLAW_PLUGIN_DATA_DIR,
process.env.HARBORFORGE_PLUGIN_DIR,
join(process.cwd(), '.harborforge'),
join(process.cwd(), 'data'),
'/tmp/harborforge',
];
for (const dir of candidates) {
if (dir) {
try {
if (!existsSync(dir)) {
mkdirSync(dir, { recursive: true });
}
// Test write access
const testFile = join(dir, '.write-test');
writeFileSync(testFile, '', { flag: 'w' });
return dir;
} catch {
continue;
}
}
}
// Fallback to current working directory
return process.cwd();
}
/**
* Persist current state to disk for recovery after restart.
*/
private persistState(reason: string): void {
try {
const persistedState: PersistedState = {
version: STATE_VERSION,
persistedAt: new Date().toISOString(),
reason,
currentSlot: this.state.currentSlot,
deferredSlotIds: Array.from(this.state.deferredSlotIds),
isProcessing: this.state.isProcessing,
agentStatus: null, // Will be determined at restore time
};
writeFileSync(this.stateFilePath, JSON.stringify(persistedState, null, 2));
this.config.logger.info(`[PLG-CAL-004] State persisted to ${this.stateFilePath} (reason: ${reason})`);
} catch (err) {
this.config.logger.error('[PLG-CAL-004] Failed to persist state:', err);
}
}
/**
* Restore state from disk if available.
*/
private restoreState(): void {
try {
if (!existsSync(this.stateFilePath)) {
return;
}
const data = readFileSync(this.stateFilePath, 'utf-8');
const persisted: PersistedState = JSON.parse(data);
// Validate version
if (persisted.version !== STATE_VERSION) {
this.config.logger.warn(`[PLG-CAL-004] State version mismatch: ${persisted.version} vs ${STATE_VERSION}`);
this.clearPersistedState();
return;
}
// Restore deferred slot IDs
if (persisted.deferredSlotIds && persisted.deferredSlotIds.length > 0) {
this.state.deferredSlotIds = new Set(persisted.deferredSlotIds);
this.config.logger.info(`[PLG-CAL-004] Restored ${persisted.deferredSlotIds.length} deferred slot(s)`);
}
// If there was a slot in progress, mark it for replanning
if (persisted.isProcessing && persisted.currentSlot) {
this.config.logger.warn(
`[PLG-CAL-004] Previous session had in-progress slot: ${this.getSlotId(persisted.currentSlot)}`
);
// The slot will be picked up by the next heartbeat and can be resumed or deferred
}
this.config.logger.info(`[PLG-CAL-004] State restored from ${persisted.persistedAt} (reason: ${persisted.reason})`);
// Clear the persisted state after successful restore
this.clearPersistedState();
} catch (err) {
this.config.logger.error('[PLG-CAL-004] Failed to restore state:', err);
}
}
/**
* Clear persisted state file after successful restore.
*/
private clearPersistedState(): void {
try {
if (existsSync(this.stateFilePath)) {
// In a real implementation, we might want to archive instead of delete
// For now, we'll just clear the content to mark as processed
writeFileSync(this.stateFilePath, JSON.stringify({ restored: true, at: new Date().toISOString() }));
}
} catch (err) {
this.config.logger.error('[PLG-CAL-004] Failed to clear persisted state:', err);
}
}
/**
* Send a final heartbeat to the backend before shutdown.
*/
private async sendFinalHeartbeat(reason: string): Promise<void> {
try {
this.config.logger.info(`[PLG-CAL-004] Sending final heartbeat (reason: ${reason})`);
// Send agent status update indicating we're going offline
await this.config.bridge.reportAgentStatus({ status: 'offline' });
this.config.logger.info('[PLG-CAL-004] Final heartbeat sent successfully');
} catch (err) {
this.config.logger.error('[PLG-CAL-004] Failed to send final heartbeat:', err);
}
}
/**
* Handle ScheduledGatewayRestart event.
* PLG-CAL-004: Persist state, send final heartbeat, pause scheduled tasks.
*/
private async handleScheduledGatewayRestart(slot: CalendarSlotResponse): Promise<void> {
this.config.logger.info('[PLG-CAL-004] Handling ScheduledGatewayRestart event');
// 1. Mark restart as pending to prevent new slot processing
this.state.isRestartPending = true;
// 2. Persist current state
this.persistState('ScheduledGatewayRestart');
// 3. If there's a current slot, pause it gracefully
if (this.state.isProcessing && this.state.currentSlot) {
this.config.logger.info('[PLG-CAL-004] Pausing current slot before restart');
await this.pauseCurrentSlot();
}
// 4. Send final heartbeat
await this.sendFinalHeartbeat('ScheduledGatewayRestart');
// 5. Stop the scheduler (pause scheduled tasks)
this.config.logger.info('[PLG-CAL-004] Stopping scheduler due to gateway restart');
this.stop();
// 6. Mark the slot as finished (since we've handled the restart)
const update: SlotAgentUpdate = {
status: SlotStatus.FINISHED,
actual_duration: 0, // Restart preparation doesn't take time
};
try {
if (slot.id) {
await this.config.bridge.updateSlot(slot.id, update);
} else if (slot.virtual_id) {
await this.config.bridge.updateVirtualSlot(slot.virtual_id, update);
}
} catch (err) {
this.config.logger.error('[PLG-CAL-004] Failed to mark restart slot as finished:', err);
}
}
/**
@@ -112,6 +324,7 @@ export class CalendarScheduler {
}
this.state.isRunning = true;
this.state.isRestartPending = false;
this.config.logger.info('Calendar scheduler started');
// Run initial heartbeat immediately
@@ -148,6 +361,12 @@ export class CalendarScheduler {
return;
}
// Skip heartbeat if restart is pending
if (this.state.isRestartPending) {
this.logDebug('Heartbeat skipped: gateway restart pending');
return;
}
this.state.lastHeartbeatAt = new Date();
try {
@@ -159,7 +378,9 @@ export class CalendarScheduler {
return;
}
this.logDebug(`Heartbeat: ${response.slots.length} slots pending, agent_status=${response.agent_status}`);
this.logDebug(
`Heartbeat: ${response.slots.length} slots pending, agent_status=${response.agent_status}`
);
// If agent is not idle, defer all pending slots
if (response.agent_status !== 'idle') {
@@ -178,7 +399,6 @@ export class CalendarScheduler {
// Agent is idle - handle pending slots
await this.handleIdleAgent(response.slots);
} catch (err) {
this.config.logger.error('Heartbeat error:', err);
}
@@ -225,7 +445,7 @@ export class CalendarScheduler {
// Filter out already deferred slots in this session
const eligibleSlots = slots.filter(
s => !this.state.deferredSlotIds.has(this.getSlotId(s))
(s) => !this.state.deferredSlotIds.has(this.getSlotId(s))
);
if (eligibleSlots.length === 0) {
@@ -238,7 +458,7 @@ export class CalendarScheduler {
this.config.logger.info(
`Selected slot for execution: id=${this.getSlotId(selectedSlot)}, ` +
`type=${selectedSlot.slot_type}, priority=${selectedSlot.priority}`
`type=${selectedSlot.slot_type}, priority=${selectedSlot.priority}`
);
// Mark remaining slots as deferred
@@ -247,10 +467,27 @@ export class CalendarScheduler {
this.state.deferredSlotIds.add(this.getSlotId(slot));
}
// Check if this is a ScheduledGatewayRestart event
if (this.isScheduledGatewayRestart(selectedSlot)) {
await this.handleScheduledGatewayRestart(selectedSlot);
return;
}
// Wake agent to execute selected slot
await this.executeSlot(selectedSlot);
}
/**
* Check if a slot is a ScheduledGatewayRestart system event.
*/
private isScheduledGatewayRestart(slot: CalendarSlotResponse): boolean {
if (slot.event_type !== 'system_event' || !slot.event_data) {
return false;
}
const sysData = slot.event_data as CalendarEventDataSystemEvent;
return sysData.event === 'ScheduledGatewayRestart';
}
/**
* Execute a slot by waking the agent.
*/
@@ -315,7 +552,6 @@ export class CalendarScheduler {
// Note: isProcessing remains true until agent signals completion
// This is handled by external completion callback
} catch (err) {
this.config.logger.error('Error executing slot:', err);
this.state.isProcessing = false;
@@ -361,7 +597,10 @@ export class CalendarScheduler {
/**
* Build prompt for job-type slots.
*/
private buildJobPrompt(slot: CalendarSlotResponse, jobData: CalendarEventDataJob): string {
private buildJobPrompt(
slot: CalendarSlotResponse,
jobData: CalendarEventDataJob
): string {
const duration = slot.estimated_duration;
const type = jobData.type;
const code = jobData.code;
@@ -518,7 +757,6 @@ Please use this time for the scheduled activity.`;
this.config.logger.info(
`Completed slot ${this.getSlotId(slot)}, actual_duration=${actualDurationMinutes}min`
);
} catch (err) {
this.config.logger.error('Failed to complete slot:', err);
} finally {
@@ -556,7 +794,6 @@ Please use this time for the scheduled activity.`;
this.config.logger.info(
`Aborted slot ${this.getSlotId(slot)}${reason ? `: ${reason}` : ''}`
);
} catch (err) {
this.config.logger.error('Failed to abort slot:', err);
} finally {
@@ -589,7 +826,6 @@ Please use this time for the scheduled activity.`;
}
this.config.logger.info(`Paused slot ${this.getSlotId(slot)}`);
} catch (err) {
this.config.logger.error('Failed to pause slot:', err);
}
@@ -617,7 +853,6 @@ Please use this time for the scheduled activity.`;
}
this.config.logger.info(`Resumed slot ${this.getSlotId(slot)}`);
} catch (err) {
this.config.logger.error('Failed to resume slot:', err);
}
@@ -692,6 +927,20 @@ Please use this time for the scheduled activity.`;
getCurrentSlot(): CalendarSlotResponse | null {
return this.state.currentSlot;
}
/**
* Check if a gateway restart is pending.
*/
isRestartPending(): boolean {
return this.state.isRestartPending;
}
/**
* Get the path to the state file.
*/
getStateFilePath(): string {
return this.stateFilePath;
}
}
/**

71
plugin/index.ts
View File

@@ -5,7 +5,7 @@
* for the HarborForge Monitor bridge (via monitor_port).
*
* Also integrates with HarborForge Calendar system to wake agents
* for scheduled tasks (PLG-CAL-002).
* for scheduled tasks (PLG-CAL-002, PLG-CAL-004).
*
* Sidecar architecture has been removed. Telemetry data is now
* served directly by the plugin when Monitor queries via the
@@ -105,7 +105,7 @@ export default {
},
openclaw: {
version: api.version || 'unknown',
pluginVersion: '0.3.0',
pluginVersion: '0.3.1', // Bumped for PLG-CAL-004
},
timestamp: new Date().toISOString(),
};
@@ -128,7 +128,7 @@ export default {
const meta: OpenClawMeta = {
version: api.version || 'unknown',
plugin_version: '0.3.0',
plugin_version: '0.3.1',
agents: [], // TODO: populate from api agent list when available
};
@@ -363,6 +363,7 @@ export default {
running: calendarScheduler.isRunning(),
processing: calendarScheduler.isProcessing(),
currentSlot: calendarScheduler.getCurrentSlot(),
isRestartPending: calendarScheduler.isRestartPending(),
} : null;
return {
@@ -439,6 +440,8 @@ export default {
processing: calendarScheduler.isProcessing(),
currentSlot: calendarScheduler.getCurrentSlot(),
state: calendarScheduler.getState(),
isRestartPending: calendarScheduler.isRestartPending(),
stateFilePath: calendarScheduler.getStateFilePath(),
};
},
}));
@@ -490,6 +493,68 @@ export default {
},
}));
// Tool: pause current slot
api.registerTool(() => ({
name: 'harborforge_calendar_pause',
description: 'Pause the current calendar slot',
parameters: {
type: 'object',
properties: {},
},
async execute() {
if (!calendarScheduler) {
return { error: 'Calendar scheduler not running' };
}
await calendarScheduler.pauseCurrentSlot();
return { success: true, message: 'Slot paused' };
},
}));
// Tool: resume current slot
api.registerTool(() => ({
name: 'harborforge_calendar_resume',
description: 'Resume the paused calendar slot',
parameters: {
type: 'object',
properties: {},
},
async execute() {
if (!calendarScheduler) {
return { error: 'Calendar scheduler not running' };
}
await calendarScheduler.resumeCurrentSlot();
return { success: true, message: 'Slot resumed' };
},
}));
// Tool: check ScheduledGatewayRestart status
api.registerTool(() => ({
name: 'harborforge_restart_status',
description: 'Check if a gateway restart is pending (PLG-CAL-004)',
parameters: {
type: 'object',
properties: {},
},
async execute() {
if (!calendarScheduler) {
return { error: 'Calendar scheduler not running' };
}
const isPending = calendarScheduler.isRestartPending();
const stateFilePath = calendarScheduler.getStateFilePath();
return {
isRestartPending: isPending,
stateFilePath: stateFilePath,
message: isPending
? 'A gateway restart has been scheduled. The scheduler has been paused.'
: 'No gateway restart is pending.',
};
},
}));
logger.info('HarborForge plugin registered (id: harbor-forge)');
},
};