From 3c61cab44e8a36b5b1141661e326032844594428 Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Sun, 7 Dec 2025 21:42:54 -0700 Subject: [PATCH 01/25] Finalized Wakeup Core Schema Added Route Comments to working files. This allows easier use with website-based chat agents. Due to them naming files the same thing. EX: Claude will named them just index.ts index.ts. This allows you to open it and read the top and see what file it is without looking at the actual file. scheduled_wakeups --- AGENTS.md | 37 +++++ packages/kernl/src/index.ts | 4 + packages/storage/core/src/memory/index.ts | 1 + packages/storage/core/src/memory/schema.ts | 1 + packages/storage/core/src/table.ts | 1 + packages/storage/core/src/wakeup/index.ts | 10 ++ packages/storage/core/src/wakeup/schema.ts | 65 +++++++++ packages/storage/pg/src/migrations.ts | 7 + packages/storage/pg/src/storage.ts | 3 + packages/storage/pg/src/wakeup/store.ts | 154 +++++++++++++++++++++ 10 files changed, 283 insertions(+) create mode 100644 AGENTS.md create mode 100644 packages/storage/core/src/wakeup/index.ts create mode 100644 packages/storage/core/src/wakeup/schema.ts create mode 100644 packages/storage/pg/src/wakeup/store.ts diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 00000000..291ed7ca --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,37 @@ +# Project Instructions for Claude Code + +## Development Guidelines + +- We use **pnpm** as our package manager +- At the beginning of any conversation, familiarize yourself with the repo structure +- Prefer method names with linux style, lowercase short words +- Never promote yourself in commits (no "Generated with Claude" messages) +- If we refactor something and are no longer using an old file, clean up after yourself + +`pnpm build` is the build command - not `pnpm build:_` or anything else. + +--- + +# Rule: Read the Docs First + +**You must do this before proposing designs, writing code, or editing files:** + +## 1) Read the canonical specs in `/docs/` +Review the relevant files and design constraints before you suggest anything. + +## 2) Follow established codebase patterns +- **Examine existing code** to understand established patterns, conventions, and architectural decisions +- **Maintain consistency** with existing naming conventions, file structures, and coding patterns +- **Reuse existing patterns** rather than inventing new ones unless the spec explicitly requires deviation +- If you must deviate from existing patterns, explicitly justify why based on the `/docs/` specifications + +## 3) If the spec is missing or ambiguous +- **Stop and propose a spec addition** first (succinct ADR-style note). Do not implement until the spec gap is resolved. +- Offer a default, but mark it as **"Proposed addition to /docs"**. + +## Checklists + +### Before coding +- [ ] I looked up the relevant `/docs/*` file(s). +- [ ] I examined existing codebase patterns for similar functionality. +- [ ] I verified my approach is consistent with established conventions. diff --git a/packages/kernl/src/index.ts b/packages/kernl/src/index.ts index 109df52f..963b1428 100644 --- a/packages/kernl/src/index.ts +++ b/packages/kernl/src/index.ts @@ -1,3 +1,7 @@ +/** + * + * /package/kernl/src/index.ts +*/ export { Kernl } from "./kernl"; export type { KernlOptions, diff --git a/packages/storage/core/src/memory/index.ts b/packages/storage/core/src/memory/index.ts index 294ff8ff..a12e4367 100644 --- a/packages/storage/core/src/memory/index.ts +++ b/packages/storage/core/src/memory/index.ts @@ -1,5 +1,6 @@ /** * Memory storage types and schema. + * /packages/storage/core/src/memory/index.ts */ export { diff --git a/packages/storage/core/src/memory/schema.ts b/packages/storage/core/src/memory/schema.ts index 8da61f0e..210f8ca9 100644 --- a/packages/storage/core/src/memory/schema.ts +++ b/packages/storage/core/src/memory/schema.ts @@ -1,5 +1,6 @@ /** * Memory table definition and record schema. + * /packages/storage/core/src/memory/schema.ts */ import { z } from "zod"; diff --git a/packages/storage/core/src/table.ts b/packages/storage/core/src/table.ts index 437fba3d..805088ec 100644 --- a/packages/storage/core/src/table.ts +++ b/packages/storage/core/src/table.ts @@ -1,5 +1,6 @@ /** * Schema name for all kernl tables. + * /packages/storage/core/table.ts */ export const KERNL_SCHEMA_NAME = "kernl"; diff --git a/packages/storage/core/src/wakeup/index.ts b/packages/storage/core/src/wakeup/index.ts new file mode 100644 index 00000000..68c9fcaf --- /dev/null +++ b/packages/storage/core/src/wakeup/index.ts @@ -0,0 +1,10 @@ +/** + * Wakeup storage types and utilities. + * /packages/storage/core/src/wakeup/index.ts + */ + +export { + TABLE_SCHEDULED_WAKEUPS, + ScheduledWakeupRecordSchema, + type ScheduledWakeupRecord, +} from "./schema"; diff --git a/packages/storage/core/src/wakeup/schema.ts b/packages/storage/core/src/wakeup/schema.ts new file mode 100644 index 00000000..bd53b8ed --- /dev/null +++ b/packages/storage/core/src/wakeup/schema.ts @@ -0,0 +1,65 @@ +/** + * /packages/storage/core/src/wakeup/schema.ts + * + * First implementation: + * - run_at_s: epoch seconds when the wakeup becomes due + * - woken: consumed/completed + * - claimed_at_s: set when a poller claims it to avoid double-processing + */ + +import { z } from "zod"; + +import { text, bigint, boolean, timestamps, defineTable } from "@/table"; +import { TABLE_THREADS } from "@/thread/schema"; + +export const TABLE_SCHEDULED_WAKEUPS = defineTable( + "scheduled_wakeups", + { + id: text().primaryKey(), + + thread_id: text().references(() => TABLE_THREADS.columns.id, { + onDelete: "CASCADE", + }), + + // Due time (epoch seconds) + run_at_s: bigint(), + + reason: text().nullable(), + + // Consumed/completed + woken: boolean().default(false), + + // Claimed by a poller (epoch seconds); nullable means unclaimed + claimed_at_s: bigint().nullable(), + + ...timestamps, + + error: text().nullable(), + }, + [ + // Polling query: woken=false AND claimed_at_s IS NULL AND run_at_s <= now + { kind: "index", columns: ["woken", "run_at_s"] }, + { kind: "index", columns: ["thread_id"] }, + { kind: "index", columns: ["claimed_at_s"] }, + ], +); + +const epochSeconds = z.coerce.number().int().nonnegative(); + +export const ScheduledWakeupRecordSchema = z.object({ + id: z.string(), + thread_id: z.string(), + + run_at_s: epochSeconds, + reason: z.string().nullable(), + + woken: z.boolean(), + claimed_at_s: epochSeconds.nullable(), + + created_at: epochSeconds, + updated_at: epochSeconds, + + error: z.string().nullable(), +}); + +export type ScheduledWakeupRecord = z.infer; diff --git a/packages/storage/pg/src/migrations.ts b/packages/storage/pg/src/migrations.ts index 77416325..07e239d0 100644 --- a/packages/storage/pg/src/migrations.ts +++ b/packages/storage/pg/src/migrations.ts @@ -8,6 +8,7 @@ import { TABLE_THREADS, TABLE_THREAD_EVENTS, TABLE_MEMORIES, + TABLE_SCHEDULED_WAKEUPS, } from "@kernl-sdk/storage"; /** @@ -40,6 +41,12 @@ export const MIGRATIONS: Migration[] = [ await ctx.createTable(TABLE_MEMORIES); }, }, + { + id: "003_scheduled_wakeups", + async up(ctx) { + await ctx.createTable(TABLE_SCHEDULED_WAKEUPS); + }, + }, ]; /** diff --git a/packages/storage/pg/src/storage.ts b/packages/storage/pg/src/storage.ts index b7cc21d7..9292e78c 100644 --- a/packages/storage/pg/src/storage.ts +++ b/packages/storage/pg/src/storage.ts @@ -15,6 +15,7 @@ import { UnimplementedError } from "@kernl-sdk/shared/lib"; /* pg */ import { PGThreadStore } from "./thread/store"; import { PGMemoryStore } from "./memory/store"; +import { PGWakeupStore } from "./wakeup/store"; import { MIGRATIONS } from "./migrations"; import { SQL_IDENTIFIER_REGEX } from "./sql"; @@ -109,11 +110,13 @@ export class PGStorage implements KernlStorage { threads: PGThreadStore; memories: PGMemoryStore; + wakeups: PGWakeupStore; constructor(config: PGStorageConfig) { this.pool = config.pool; this.threads = new PGThreadStore(this.pool, () => this.ensureInit()); this.memories = new PGMemoryStore(this.pool, () => this.ensureInit()); + this.wakeups = new PGWakeupStore(this.pool, () => this.ensureInit()); } /** diff --git a/packages/storage/pg/src/wakeup/store.ts b/packages/storage/pg/src/wakeup/store.ts new file mode 100644 index 00000000..cd8dc45b --- /dev/null +++ b/packages/storage/pg/src/wakeup/store.ts @@ -0,0 +1,154 @@ +import type { Pool, PoolClient } from "pg"; + +import type { + WakeupStore, + NewScheduledWakeup, + ScheduledWakeup, + ScheduledWakeupUpdate, +} from "kernl"; +import { + KERNL_SCHEMA_NAME, + NewScheduledWakeupCodec, + ScheduledWakeupRecordCodec, + type ScheduledWakeupRecord, +} from "@kernl-sdk/storage"; + +/** + * PostgreSQL wakeup store implementation. + */ +export class PGWakeupStore implements WakeupStore { + private db: Pool | PoolClient; + private ensureInit: () => Promise; + + constructor(db: Pool | PoolClient, ensureInit: () => Promise) { + this.db = db; + this.ensureInit = ensureInit; + } + + /** + * Create a scheduled wakeup. + */ + async create(wakeup: NewScheduledWakeup): Promise { + await this.ensureInit(); + const record = NewScheduledWakeupCodec.encode(wakeup); + + const result = await this.db.query( + `INSERT INTO ${KERNL_SCHEMA_NAME}.scheduled_wakeups + (id, thread_id, wait_time_ms, reason, woken, claimed_at, created_at, updated_at, error) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + RETURNING *`, + [ + record.id, + record.thread_id, + record.wait_time_ms, + record.reason, + record.woken, + record.claimed_at, + record.created_at, + record.updated_at, + record.error, + ], + ); + + return ScheduledWakeupRecordCodec.decode(result.rows[0]); + } + + /** + * Atomically claim due wakeups using SKIP LOCKED to avoid double-processing. + */ + async claimDue(limit: number): Promise { + await this.ensureInit(); + if (limit <= 0) return []; + + const now = Date.now(); + + const result = await this.db.query( + ` + WITH due AS ( + SELECT id + FROM ${KERNL_SCHEMA_NAME}.scheduled_wakeups + WHERE woken = false + AND claimed_at IS NULL + AND created_at + wait_time_ms <= $1::bigint + ORDER BY created_at ASC + LIMIT $2 + FOR UPDATE SKIP LOCKED + ) + UPDATE ${KERNL_SCHEMA_NAME}.scheduled_wakeups w + SET claimed_at = $1, + updated_at = $1 + FROM due + WHERE w.id = due.id + RETURNING w.* + `, + [now, limit], + ); + + return result.rows.map((row) => ScheduledWakeupRecordCodec.decode(row)); + } + + /** + * Update wakeup status/error fields. + */ + async update( + id: string, + patch: ScheduledWakeupUpdate, + ): Promise { + await this.ensureInit(); + + const fields: string[] = []; + const params: any[] = []; + let idx = 1; + + if (patch.woken !== undefined) { + fields.push(`woken = $${idx++}`); + params.push(patch.woken); + } + if (patch.claimedAt !== undefined) { + fields.push(`claimed_at = $${idx++}`); + params.push(patch.claimedAt ? patch.claimedAt.getTime() : null); + } + + if (patch.error !== undefined) { + fields.push(`error = $${idx++}`); + params.push(patch.error); + } + + const updatedAt = patch.updatedAt?.getTime() ?? Date.now(); + fields.push(`updated_at = $${idx++}`); + params.push(updatedAt); + + params.push(id); + const idParam = idx; + + const result = await this.db.query( + ` + UPDATE ${KERNL_SCHEMA_NAME}.scheduled_wakeups + SET ${fields.join(", ")} + WHERE id = $${idParam} + RETURNING * + `, + params, + ); + + if (result.rows.length === 0) { + throw new Error(`Wakeup ${id} not found`); + } + + return ScheduledWakeupRecordCodec.decode(result.rows[0]); + } + + /** + * Cancel all pending wakeups for a thread. + */ + async cancelForThread(tid: string): Promise { + await this.ensureInit(); + await this.db.query( + ` + DELETE FROM ${KERNL_SCHEMA_NAME}.scheduled_wakeups + WHERE thread_id = $1 AND woken = false + `, + [tid], + ); + } +} From 6fad733f791d557b0f7ae021f251640aea45c89c Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Mon, 8 Dec 2025 09:22:39 -0700 Subject: [PATCH 02/25] Added Routes to working files. Added routes to working files so it's easier to chat with. --- packages/kernl/src/context.ts | 4 ++++ packages/kernl/src/index.ts | 1 - packages/kernl/src/memory/codecs/domain.ts | 1 + packages/kernl/src/memory/codecs/identity.ts | 1 + packages/kernl/src/memory/codecs/index.ts | 1 + packages/kernl/src/memory/codecs/tpuf.ts | 1 + packages/kernl/src/memory/encoder.ts | 1 + packages/kernl/src/memory/handle.ts | 1 + packages/kernl/src/memory/index.ts | 1 + packages/kernl/src/memory/indexes.ts | 1 + packages/kernl/src/memory/memory.ts | 1 + packages/kernl/src/memory/schema.ts | 1 + packages/kernl/src/memory/store.ts | 1 + packages/kernl/src/memory/types.ts | 1 + packages/kernl/src/storage/index.ts | 1 + packages/kernl/src/thread/index.ts | 4 ++++ packages/kernl/src/thread/thread.ts | 3 +++ packages/kernl/src/thread/types.ts | 3 +++ packages/kernl/src/thread/utils.ts | 3 +++ packages/kernl/src/tool/index.ts | 4 ++++ packages/kernl/src/tool/sys/index.ts | 3 +++ packages/kernl/src/tool/sys/memory.ts | 1 + packages/kernl/src/tool/tool.ts | 4 ++++ packages/kernl/src/tool/toolkit.ts | 4 ++++ packages/kernl/src/tool/types.ts | 4 ++++ packages/shared/src/lib/codec.ts | 1 + packages/storage/pg/src/memory/sql.ts | 3 ++- packages/storage/pg/src/memory/store.ts | 1 + 28 files changed, 54 insertions(+), 2 deletions(-) diff --git a/packages/kernl/src/context.ts b/packages/kernl/src/context.ts index edecab7d..f4c3be4a 100644 --- a/packages/kernl/src/context.ts +++ b/packages/kernl/src/context.ts @@ -1,3 +1,7 @@ +/** + * /packages/kernl/src/context.ts + */ + import type { Agent } from "@/agent"; /** diff --git a/packages/kernl/src/index.ts b/packages/kernl/src/index.ts index 963b1428..ef550364 100644 --- a/packages/kernl/src/index.ts +++ b/packages/kernl/src/index.ts @@ -1,5 +1,4 @@ /** - * * /package/kernl/src/index.ts */ export { Kernl } from "./kernl"; diff --git a/packages/kernl/src/memory/codecs/domain.ts b/packages/kernl/src/memory/codecs/domain.ts index 24207ebc..0caf7a40 100644 --- a/packages/kernl/src/memory/codecs/domain.ts +++ b/packages/kernl/src/memory/codecs/domain.ts @@ -1,5 +1,6 @@ /** * Domain-level memory codecs. + * /packages/kernl/src/memory/codecs/domain.ts * * Codecs for transforming between memory domain types and search/index formats. */ diff --git a/packages/kernl/src/memory/codecs/identity.ts b/packages/kernl/src/memory/codecs/identity.ts index d9c6be66..2e4ba738 100644 --- a/packages/kernl/src/memory/codecs/identity.ts +++ b/packages/kernl/src/memory/codecs/identity.ts @@ -1,5 +1,6 @@ /** * Identity codecs - pass through unchanged. + * /packages/kernl/src/memory/codecs/identity.ts * * Used for backends that support the full IndexMemoryRecord schema natively. */ diff --git a/packages/kernl/src/memory/codecs/index.ts b/packages/kernl/src/memory/codecs/index.ts index df4bd6a1..02792688 100644 --- a/packages/kernl/src/memory/codecs/index.ts +++ b/packages/kernl/src/memory/codecs/index.ts @@ -1,5 +1,6 @@ /** * Memory codecs. + * /packages/kernl/src/memory/codecs/index.ts * * Re-exports all memory codecs: * - Domain codecs (MEMORY_FILTER, PATCH_CODEC, recordCodec) diff --git a/packages/kernl/src/memory/codecs/tpuf.ts b/packages/kernl/src/memory/codecs/tpuf.ts index fb7d4c0b..d3975286 100644 --- a/packages/kernl/src/memory/codecs/tpuf.ts +++ b/packages/kernl/src/memory/codecs/tpuf.ts @@ -1,5 +1,6 @@ /** * Turbopuffer backend codecs. + * /packages/kernl/src/memory/codecs/tpuf.ts * * Turbopuffer constraints: * - Exactly one ANN vector field named "vector" per namespace. diff --git a/packages/kernl/src/memory/encoder.ts b/packages/kernl/src/memory/encoder.ts index 35398933..736b170e 100644 --- a/packages/kernl/src/memory/encoder.ts +++ b/packages/kernl/src/memory/encoder.ts @@ -1,5 +1,6 @@ /** * MemoryByte encoder - converts MemoryByte to IndexableByte with embeddings. + * /packages/kernl/src/memory/encoder.ts */ import type { EmbeddingModel, JSONObject } from "@kernl-sdk/protocol"; diff --git a/packages/kernl/src/memory/handle.ts b/packages/kernl/src/memory/handle.ts index 765ba3aa..d08e9b60 100644 --- a/packages/kernl/src/memory/handle.ts +++ b/packages/kernl/src/memory/handle.ts @@ -1,5 +1,6 @@ /** * Memory index handle with lazy initialization. + * /packages/kernl/src/memory/handle.ts */ import { diff --git a/packages/kernl/src/memory/index.ts b/packages/kernl/src/memory/index.ts index f2ec7707..48c75509 100644 --- a/packages/kernl/src/memory/index.ts +++ b/packages/kernl/src/memory/index.ts @@ -1,5 +1,6 @@ /** * Memory module. + * /packages/kernl/src/memory/index.ts */ export { Memory } from "./memory"; diff --git a/packages/kernl/src/memory/indexes.ts b/packages/kernl/src/memory/indexes.ts index aafd0503..0004a765 100644 --- a/packages/kernl/src/memory/indexes.ts +++ b/packages/kernl/src/memory/indexes.ts @@ -1,5 +1,6 @@ /** * Memory index interfaces. + * /packages/kernl/src/memory/indexes.ts * * Indexes are projections of the primary store (DB) that enable * specialized query patterns (vector search, graph traversal, archival). diff --git a/packages/kernl/src/memory/memory.ts b/packages/kernl/src/memory/memory.ts index d29f2783..a4ab996b 100644 --- a/packages/kernl/src/memory/memory.ts +++ b/packages/kernl/src/memory/memory.ts @@ -20,6 +20,7 @@ import { MEMORY_FILTER, PATCH_CODEC, recordCodec } from "./codecs"; /** * Memory is the primary memory abstraction for agents. + * /packages/kernl/src/memory/memory.ts * * Sits above storage/index layers + owns cognitive policy, eviction/TTL, consolidation. * diff --git a/packages/kernl/src/memory/schema.ts b/packages/kernl/src/memory/schema.ts index 4308589d..e451d9f9 100644 --- a/packages/kernl/src/memory/schema.ts +++ b/packages/kernl/src/memory/schema.ts @@ -1,5 +1,6 @@ /** * Memory index schema builder. + * /packages/kernl/src/memory/schema.ts */ import type { FieldSchema } from "@kernl-sdk/retrieval"; diff --git a/packages/kernl/src/memory/store.ts b/packages/kernl/src/memory/store.ts index 62466ec8..daf6583c 100644 --- a/packages/kernl/src/memory/store.ts +++ b/packages/kernl/src/memory/store.ts @@ -1,5 +1,6 @@ /** * Memory store interface. + * /packages/kernl/src/memory/store.ts */ import type { diff --git a/packages/kernl/src/memory/types.ts b/packages/kernl/src/memory/types.ts index 3f573afa..137101ed 100644 --- a/packages/kernl/src/memory/types.ts +++ b/packages/kernl/src/memory/types.ts @@ -1,5 +1,6 @@ /** * Memory types. + * /packages/kernl/src/memory/types.ts */ import type { JSONObject } from "@kernl-sdk/protocol"; diff --git a/packages/kernl/src/storage/index.ts b/packages/kernl/src/storage/index.ts index e4aecdba..2481f1ec 100644 --- a/packages/kernl/src/storage/index.ts +++ b/packages/kernl/src/storage/index.ts @@ -1,5 +1,6 @@ /** * Storage contracts for Kernl. + * /packages/kernl/src/storage/index.ts * * Core owns these interfaces; storage packages implement them. * (must be defined here to avoid circular deps) diff --git a/packages/kernl/src/thread/index.ts b/packages/kernl/src/thread/index.ts index 45388ec2..9ac6996f 100644 --- a/packages/kernl/src/thread/index.ts +++ b/packages/kernl/src/thread/index.ts @@ -1 +1,5 @@ +/** + * /packages/kernl/src/thread/index.ts + */ + export { Thread } from "./thread"; diff --git a/packages/kernl/src/thread/thread.ts b/packages/kernl/src/thread/thread.ts index 21226e75..7d2302cc 100644 --- a/packages/kernl/src/thread/thread.ts +++ b/packages/kernl/src/thread/thread.ts @@ -1,3 +1,6 @@ +/** + * /packages/kernl/src/thread/thread.ts + */ import assert from "assert"; import { ZodType } from "zod"; import * as z from "zod"; diff --git a/packages/kernl/src/thread/types.ts b/packages/kernl/src/thread/types.ts index 2f898c51..faeedd33 100644 --- a/packages/kernl/src/thread/types.ts +++ b/packages/kernl/src/thread/types.ts @@ -1,3 +1,6 @@ +/** + * /packages/kernl/src/thread/types.ts + */ import { ToolCall, LanguageModel, diff --git a/packages/kernl/src/thread/utils.ts b/packages/kernl/src/thread/utils.ts index 45b53e5a..707c6e4c 100644 --- a/packages/kernl/src/thread/utils.ts +++ b/packages/kernl/src/thread/utils.ts @@ -1,3 +1,6 @@ +/** + * /packages/kernl/src/thread/utils.ts + */ import { ZodType } from "zod"; import type { ResolvedAgentResponse } from "@/guardrail"; diff --git a/packages/kernl/src/tool/index.ts b/packages/kernl/src/tool/index.ts index 2dba7f93..3c5488b0 100644 --- a/packages/kernl/src/tool/index.ts +++ b/packages/kernl/src/tool/index.ts @@ -1,3 +1,7 @@ +/** + * /packages/kernl/src/tool/index.ts + */ + export { BaseTool, FunctionTool, HostedTool, tool } from "./tool"; export { BaseToolkit, Toolkit, FunctionToolkit, MCPToolkit } from "./toolkit"; export type { diff --git a/packages/kernl/src/tool/sys/index.ts b/packages/kernl/src/tool/sys/index.ts index a2da55a2..fc1a7db0 100644 --- a/packages/kernl/src/tool/sys/index.ts +++ b/packages/kernl/src/tool/sys/index.ts @@ -1,7 +1,10 @@ /** * System toolkits. + * /packages/kernl/src/tool/sys/index.ts * * These are internal toolkits that can be enabled via agent config flags. */ + + export { memory } from "./memory"; diff --git a/packages/kernl/src/tool/sys/memory.ts b/packages/kernl/src/tool/sys/memory.ts index 80b7c991..9693cb16 100644 --- a/packages/kernl/src/tool/sys/memory.ts +++ b/packages/kernl/src/tool/sys/memory.ts @@ -1,5 +1,6 @@ /** * Memory system toolkit. + * /packages/kernl/src/tool/sys/memory.ts * * Provides tools for agents to store and retrieve memories. * Enabled via `memory: true` in agent config. diff --git a/packages/kernl/src/tool/tool.ts b/packages/kernl/src/tool/tool.ts index dfd5bf23..83909224 100644 --- a/packages/kernl/src/tool/tool.ts +++ b/packages/kernl/src/tool/tool.ts @@ -1,3 +1,7 @@ +/** + * /packages/kernl/src/tool/tool.ts + */ + import { z } from "zod"; import { Context, UnknownContext } from "@/context"; diff --git a/packages/kernl/src/tool/toolkit.ts b/packages/kernl/src/tool/toolkit.ts index 6eed0c81..de1c5854 100644 --- a/packages/kernl/src/tool/toolkit.ts +++ b/packages/kernl/src/tool/toolkit.ts @@ -1,3 +1,7 @@ +/** + * /packages/kernl/src/tool/toolkit.ts + */ + import type { Agent } from "@/agent"; import type { Context, UnknownContext } from "@/context"; diff --git a/packages/kernl/src/tool/types.ts b/packages/kernl/src/tool/types.ts index 475b87e7..bbf99d40 100644 --- a/packages/kernl/src/tool/types.ts +++ b/packages/kernl/src/tool/types.ts @@ -1,3 +1,7 @@ +/** + * /packages/kernl/src/tool/types.ts + */ + import { z, type ZodType } from "zod"; import { Agent } from "@/agent"; diff --git a/packages/shared/src/lib/codec.ts b/packages/shared/src/lib/codec.ts index c0b0e8f8..b7367096 100644 --- a/packages/shared/src/lib/codec.ts +++ b/packages/shared/src/lib/codec.ts @@ -2,6 +2,7 @@ import { z, type ZodType } from "zod"; /** * Bidirectional codec for converting between types. + * /packages/shared/src/lib/codec.ts * * @example * ```typescript diff --git a/packages/storage/pg/src/memory/sql.ts b/packages/storage/pg/src/memory/sql.ts index 666887eb..08a00e57 100644 --- a/packages/storage/pg/src/memory/sql.ts +++ b/packages/storage/pg/src/memory/sql.ts @@ -1,6 +1,7 @@ /** * Memory SQL conversion codecs. - * + * /packages/storage/pg/src/memory/sql.ts + * * TODO: generalize object -> SQL conversion into a shared utility */ diff --git a/packages/storage/pg/src/memory/store.ts b/packages/storage/pg/src/memory/store.ts index b99bf909..70975b6e 100644 --- a/packages/storage/pg/src/memory/store.ts +++ b/packages/storage/pg/src/memory/store.ts @@ -1,5 +1,6 @@ /** * PostgreSQL Memory store implementation. + * /packages/storage/pg/src/memory/store.ts */ import type { Pool, PoolClient } from "pg"; From 56d048643a22e43d0ca26d46c088ff3348b4d03c Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Mon, 8 Dec 2025 10:47:24 -0700 Subject: [PATCH 03/25] [Builds Locally] Add scheduled wakeup store and PG implementation Introduces a new wakeup scheduling contract to kernl, including types and store interface. Adds PostgreSQL-backed implementation (PGWakeupStore) with codecs and SQL helpers for persisting, updating, and claiming scheduled wakeups. Updates package exports to expose the new API. --- packages/kernl/src/index.ts | 8 + packages/kernl/src/wakeup/index.ts | 12 ++ packages/kernl/src/wakeup/store.ts | 51 +++++++ packages/kernl/src/wakeup/types.ts | 85 +++++++++++ packages/storage/core/src/index.ts | 2 + packages/storage/pg/src/index.ts | 2 + packages/storage/pg/src/wakeup/codec.ts | 85 +++++++++++ packages/storage/pg/src/wakeup/sql.ts | 74 +++++++++ packages/storage/pg/src/wakeup/store.ts | 195 +++++++++++++----------- 9 files changed, 424 insertions(+), 90 deletions(-) create mode 100644 packages/kernl/src/wakeup/index.ts create mode 100644 packages/kernl/src/wakeup/store.ts create mode 100644 packages/kernl/src/wakeup/types.ts create mode 100644 packages/storage/pg/src/wakeup/codec.ts create mode 100644 packages/storage/pg/src/wakeup/sql.ts diff --git a/packages/kernl/src/index.ts b/packages/kernl/src/index.ts index ef550364..aeab93fe 100644 --- a/packages/kernl/src/index.ts +++ b/packages/kernl/src/index.ts @@ -82,3 +82,11 @@ export type { MemoryByte, MemoryByteCodec, } from "./memory"; + +// --- wakeups --- +export type { + WakeupStore, + NewScheduledWakeup, + ScheduledWakeup, + ScheduledWakeupUpdate, +} from "./wakeup"; diff --git a/packages/kernl/src/wakeup/index.ts b/packages/kernl/src/wakeup/index.ts new file mode 100644 index 00000000..b4de287f --- /dev/null +++ b/packages/kernl/src/wakeup/index.ts @@ -0,0 +1,12 @@ +/** + * Wakeup public API. + * /packages/kernl/src/wakeup/index.ts + */ + +export type { + NewScheduledWakeup, + ScheduledWakeup, + ScheduledWakeupUpdate, +} from "./types"; + +export type { WakeupStore } from "./store"; diff --git a/packages/kernl/src/wakeup/store.ts b/packages/kernl/src/wakeup/store.ts new file mode 100644 index 00000000..9cecf20f --- /dev/null +++ b/packages/kernl/src/wakeup/store.ts @@ -0,0 +1,51 @@ +/** + * Wakeup store contract. + * /packages/kernl/src/wakeup/store.ts + */ + +import type { + NewScheduledWakeup, + ScheduledWakeup, + ScheduledWakeupUpdate, +} from "./types"; + +/** + * Persistence contract for scheduled wakeups. + * + * Implementations live in provider packages (e.g. @kernl-sdk/storage/pg). + */ +export interface WakeupStore { + /** + * Create a new scheduled wakeup. + */ + create(wakeup: NewScheduledWakeup): Promise; + + /** + * Get a wakeup by ID. + */ + get(id: string): Promise; + + /** + * Update an existing wakeup. + */ + update(id: string, patch: ScheduledWakeupUpdate): Promise; + + /** + * Delete a wakeup (admin/cleanup). Normal flow should mark `woken = true` + * instead of deleting. + */ + delete(id: string): Promise; + + /** + * Atomically claim up to `limit` due wakeups. + * + * A wakeup is "due" when: + * - woken = false + * - claimedAt is null + * - runAt <= nowMs + * + * Implementations must be safe under multiple pollers + * (e.g. using SKIP LOCKED semantics). + */ + claimDue(nowMs: number | bigint, limit: number): Promise; +} diff --git a/packages/kernl/src/wakeup/types.ts b/packages/kernl/src/wakeup/types.ts new file mode 100644 index 00000000..32785cd0 --- /dev/null +++ b/packages/kernl/src/wakeup/types.ts @@ -0,0 +1,85 @@ +/** + * Wakeup types. + * /packages/kernl/src/wakeup/types.ts + */ + +export interface NewScheduledWakeup { + /** + * Unique wakeup ID (e.g. ulid/uuid). Generated by Kernl, not by the agent. + */ + id: string; + + /** + * Thread to resume when the wakeup becomes due. + */ + threadId: string; + + /** + * When this wakeup becomes due (epoch milliseconds). + * + * Storage providers can store this as seconds (`run_at_s`) or ms, + * codecs handle conversion. + */ + runAt: number; + + /** + * Optional free-form reason for observability. + */ + reason?: string | null; +} + +/** + * Persisted wakeup record in domain form. + */ +export interface ScheduledWakeup { + id: string; + threadId: string; + + /** + * Due time in epoch milliseconds (domain-level). + */ + runAt: number; + + reason: string | null; + + /** + * True once the wakeup has been fully consumed (thread resumed / handled). + */ + woken: boolean; + + /** + * When a poller claimed this wakeup (epoch ms). `null` if unclaimed. + */ + claimedAt: number | null; + + /** + * Creation and update timestamps (epoch ms). + */ + createdAt: number; + updatedAt: number; + + /** + * Optional error message if processing/resume failed. + */ + error: string | null; +} + +/** + * Patch payload for updating a wakeup. + */ +export interface ScheduledWakeupUpdate { + id: string; + + runAt?: number; + reason?: string | null; + + woken?: boolean; + claimedAt?: number | null; + error?: string | null; + + /** + * Optional explicit updatedAt (ms). Most PG logic will just set + * this to `Date.now()` if omitted. + */ + updatedAt?: number; +} diff --git a/packages/storage/core/src/index.ts b/packages/storage/core/src/index.ts index f6cc7b88..095cf49c 100644 --- a/packages/storage/core/src/index.ts +++ b/packages/storage/core/src/index.ts @@ -1,5 +1,6 @@ /** * @kernl-sdk/storage - Generic storage abstractions for Kernl + * /packages/storage/core/src/index.ts */ export * from "./base"; @@ -8,3 +9,4 @@ export * from "./memory"; export * from "./serde/thread"; export * from "./serde/memory"; export * from "./table"; +export * from "./wakeup"; diff --git a/packages/storage/pg/src/index.ts b/packages/storage/pg/src/index.ts index 33cffc89..05ade447 100644 --- a/packages/storage/pg/src/index.ts +++ b/packages/storage/pg/src/index.ts @@ -1,5 +1,6 @@ /** * @kernl/pg - PostgreSQL storage adapter for Kernl + * /packages/storage/pg/src/index.ts */ export { @@ -12,6 +13,7 @@ export { resolveVectorConfig, } from "./storage"; export { PGMemoryStore } from "./memory/store"; +export { PGWakeupStore } from "./wakeup/store"; export { postgres, pgvector, type PostgresConfig } from "./postgres"; export { MIGRATIONS, REQUIRED_SCHEMA_VERSION } from "./migrations"; export { diff --git a/packages/storage/pg/src/wakeup/codec.ts b/packages/storage/pg/src/wakeup/codec.ts new file mode 100644 index 00000000..a6c92c50 --- /dev/null +++ b/packages/storage/pg/src/wakeup/codec.ts @@ -0,0 +1,85 @@ +/** + * Wakeup codecs. + * /packages/storage/pg/src/wakeup/codec.ts + */ + +import type { Codec } from "@kernl-sdk/shared/lib"; + +import type { + ScheduledWakeupRecord, +} from "@kernl-sdk/storage"; + +import type { + NewScheduledWakeup, + ScheduledWakeup, +} from "kernl"; + +/** + * Convert NewScheduledWakeup (domain) -> ScheduledWakeupRecord (DB). + */ +export const NewScheduledWakeupCodec: Codec< + NewScheduledWakeup, + ScheduledWakeupRecord +> = { + encode(input) { + const nowMs = Date.now(); + const runAtS = Math.floor(input.runAt / 1000); + + return { + id: input.id, + thread_id: input.threadId, + run_at_s: runAtS, + reason: input.reason ?? null, + woken: false, + claimed_at_s: null, + created_at: nowMs, + updated_at: nowMs, + error: null, + }; + }, + + decode() { + throw new Error("NewScheduledWakeupCodec.decode not implemented"); + }, +}; + +/** + * Convert between ScheduledWakeup (domain) and ScheduledWakeupRecord (DB). + */ +export const ScheduledWakeupCodec: Codec< + ScheduledWakeup, + ScheduledWakeupRecord +> = { + encode(wakeup) { + const runAtS = Math.floor(wakeup.runAt / 1000); + const claimedAtS = + wakeup.claimedAt != null ? Math.floor(wakeup.claimedAt / 1000) : null; + + return { + id: wakeup.id, + thread_id: wakeup.threadId, + run_at_s: runAtS, + reason: wakeup.reason, + woken: wakeup.woken, + claimed_at_s: claimedAtS, + created_at: wakeup.createdAt, + updated_at: wakeup.updatedAt, + error: wakeup.error, + }; + }, + + decode(record) { + return { + id: record.id, + threadId: record.thread_id, + runAt: record.run_at_s * 1000, + reason: record.reason, + woken: record.woken, + claimedAt: + record.claimed_at_s != null ? record.claimed_at_s * 1000 : null, + createdAt: record.created_at, + updatedAt: record.updated_at, + error: record.error, + }; + }, +}; diff --git a/packages/storage/pg/src/wakeup/sql.ts b/packages/storage/pg/src/wakeup/sql.ts new file mode 100644 index 00000000..d1cb7ada --- /dev/null +++ b/packages/storage/pg/src/wakeup/sql.ts @@ -0,0 +1,74 @@ +/** + * Wakeup SQL conversion codecs. + * /packages/storage/pg/src/wakeup/sql.ts + */ + +import type { Codec } from "@kernl-sdk/shared/lib"; +import type { ScheduledWakeupUpdate } from "kernl"; + +export interface SQLClause { + sql: string; + params: unknown[]; +} + +/** + * Input for building a partial UPDATE ... SET clause. + */ +export interface PatchInput { + patch: ScheduledWakeupUpdate; + startIdx: number; +} + +/** + * Build `SET ...` clause for scheduled_wakeups updates. + * + * Mirrors the style of /packages/storage/pg/src/memory/sql.ts::PATCH + */ +export const PATCH: Codec = { + encode({ patch, startIdx }) { + const sets: string[] = []; + const params: unknown[] = []; + let idx = startIdx; + + if (patch.runAt !== undefined) { + sets.push(`run_at_s = $${idx++}`); + params.push(Math.floor(patch.runAt / 1000)); + } + + if (patch.reason !== undefined) { + sets.push(`reason = $${idx++}`); + params.push(patch.reason); + } + + if (patch.woken !== undefined) { + sets.push(`woken = $${idx++}`); + params.push(patch.woken); + } + + if (patch.claimedAt !== undefined) { + sets.push(`claimed_at_s = $${idx++}`); + params.push( + patch.claimedAt != null ? Math.floor(patch.claimedAt / 1000) : null, + ); + } + + if (patch.error !== undefined) { + sets.push(`error = $${idx++}`); + params.push(patch.error); + } + + // Always bump updated_at in ms, matching memory's PATCH behavior. + const nowMs = patch.updatedAt ?? Date.now(); + sets.push(`updated_at = $${idx++}`); + params.push(nowMs); + + return { + sql: sets.join(", "), + params, + }; + }, + + decode() { + throw new Error("PATCH.decode not implemented"); + }, +}; diff --git a/packages/storage/pg/src/wakeup/store.ts b/packages/storage/pg/src/wakeup/store.ts index cd8dc45b..fb746755 100644 --- a/packages/storage/pg/src/wakeup/store.ts +++ b/packages/storage/pg/src/wakeup/store.ts @@ -1,3 +1,8 @@ +/** + * PostgreSQL Wakeup store implementation. + * /packages/storage/pg/src/wakeup/store.ts + */ + import type { Pool, PoolClient } from "pg"; import type { @@ -6,15 +11,35 @@ import type { ScheduledWakeup, ScheduledWakeupUpdate, } from "kernl"; + import { KERNL_SCHEMA_NAME, - NewScheduledWakeupCodec, - ScheduledWakeupRecordCodec, + ScheduledWakeupRecordSchema, type ScheduledWakeupRecord, } from "@kernl-sdk/storage"; +import { NewScheduledWakeupCodec, ScheduledWakeupCodec } from "./codec"; +import { PATCH } from "./sql"; + +/** + * Convert ms/bigint ms to a JS number in ms. + */ +const toMs = (value: number | bigint): number => + typeof value === "bigint" ? Number(value) : value; + +/** + * Convert ms to epoch seconds (integer). + */ +const toSeconds = (ms: number | bigint): number => + Math.floor(toMs(ms) / 1000); + /** - * PostgreSQL wakeup store implementation. + * PostgreSQL implementation of WakeupStore. + * + * Follows the same pattern as PGMemoryStore: + * - depends on ensureInit() to create tables + * - always validates with Zod Schemas + * - converts between domain types and DB records via codecs */ export class PGWakeupStore implements WakeupStore { private db: Pool | PoolClient; @@ -25,130 +50,120 @@ export class PGWakeupStore implements WakeupStore { this.ensureInit = ensureInit; } - /** - * Create a scheduled wakeup. - */ - async create(wakeup: NewScheduledWakeup): Promise { + async create(input: NewScheduledWakeup): Promise { await this.ensureInit(); - const record = NewScheduledWakeupCodec.encode(wakeup); + + const row = NewScheduledWakeupCodec.encode(input); const result = await this.db.query( `INSERT INTO ${KERNL_SCHEMA_NAME}.scheduled_wakeups - (id, thread_id, wait_time_ms, reason, woken, claimed_at, created_at, updated_at, error) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + (id, thread_id, run_at_s, reason, woken, claimed_at_s, created_at, updated_at, error) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) RETURNING *`, [ - record.id, - record.thread_id, - record.wait_time_ms, - record.reason, - record.woken, - record.claimed_at, - record.created_at, - record.updated_at, - record.error, + row.id, + row.thread_id, + row.run_at_s, + row.reason, + row.woken, + row.claimed_at_s, + row.created_at, + row.updated_at, + row.error, ], ); - return ScheduledWakeupRecordCodec.decode(result.rows[0]); + const record = ScheduledWakeupRecordSchema.parse(result.rows[0]); + return ScheduledWakeupCodec.decode(record); } - /** - * Atomically claim due wakeups using SKIP LOCKED to avoid double-processing. - */ - async claimDue(limit: number): Promise { + async get(id: string): Promise { await this.ensureInit(); - if (limit <= 0) return []; - - const now = Date.now(); const result = await this.db.query( - ` - WITH due AS ( - SELECT id - FROM ${KERNL_SCHEMA_NAME}.scheduled_wakeups - WHERE woken = false - AND claimed_at IS NULL - AND created_at + wait_time_ms <= $1::bigint - ORDER BY created_at ASC - LIMIT $2 - FOR UPDATE SKIP LOCKED - ) - UPDATE ${KERNL_SCHEMA_NAME}.scheduled_wakeups w - SET claimed_at = $1, - updated_at = $1 - FROM due - WHERE w.id = due.id - RETURNING w.* - `, - [now, limit], + `SELECT * FROM ${KERNL_SCHEMA_NAME}.scheduled_wakeups WHERE id = $1`, + [id], ); - return result.rows.map((row) => ScheduledWakeupRecordCodec.decode(row)); + if (result.rows.length === 0) return null; + + const record = ScheduledWakeupRecordSchema.parse(result.rows[0]); + return ScheduledWakeupCodec.decode(record); } - /** - * Update wakeup status/error fields. - */ async update( id: string, patch: ScheduledWakeupUpdate, ): Promise { await this.ensureInit(); - const fields: string[] = []; - const params: any[] = []; - let idx = 1; - - if (patch.woken !== undefined) { - fields.push(`woken = $${idx++}`); - params.push(patch.woken); - } - if (patch.claimedAt !== undefined) { - fields.push(`claimed_at = $${idx++}`); - params.push(patch.claimedAt ? patch.claimedAt.getTime() : null); - } - - if (patch.error !== undefined) { - fields.push(`error = $${idx++}`); - params.push(patch.error); - } - - const updatedAt = patch.updatedAt?.getTime() ?? Date.now(); - fields.push(`updated_at = $${idx++}`); - params.push(updatedAt); - - params.push(id); - const idParam = idx; + const { sql, params } = PATCH.encode({ patch, startIdx: 2 }); const result = await this.db.query( - ` - UPDATE ${KERNL_SCHEMA_NAME}.scheduled_wakeups - SET ${fields.join(", ")} - WHERE id = $${idParam} - RETURNING * - `, - params, + `UPDATE ${KERNL_SCHEMA_NAME}.scheduled_wakeups + SET ${sql} + WHERE id = $1 + RETURNING *`, + [id, ...params], ); if (result.rows.length === 0) { - throw new Error(`Wakeup ${id} not found`); + throw new Error(`Wakeup with id ${id} not found`); } - return ScheduledWakeupRecordCodec.decode(result.rows[0]); + const record = ScheduledWakeupRecordSchema.parse(result.rows[0]); + return ScheduledWakeupCodec.decode(record); + } + + async delete(id: string): Promise { + await this.ensureInit(); + + await this.db.query( + `DELETE FROM ${KERNL_SCHEMA_NAME}.scheduled_wakeups WHERE id = $1`, + [id], + ); } /** - * Cancel all pending wakeups for a thread. + * Atomically claim up to `limit` due wakeups. + * + * Uses SELECT ... FOR UPDATE SKIP LOCKED pattern. */ - async cancelForThread(tid: string): Promise { + async claimDue( + nowMs: number | bigint, + limit: number, + ): Promise { await this.ensureInit(); - await this.db.query( + + const nowMsNum = toMs(nowMs); + const nowS = toSeconds(nowMs); + + const result = await this.db.query( ` - DELETE FROM ${KERNL_SCHEMA_NAME}.scheduled_wakeups - WHERE thread_id = $1 AND woken = false - `, - [tid], + WITH due AS ( + SELECT id + FROM ${KERNL_SCHEMA_NAME}.scheduled_wakeups + WHERE woken = FALSE + AND claimed_at_s IS NULL + AND run_at_s <= $1 + ORDER BY run_at_s ASC + LIMIT $2 + FOR UPDATE SKIP LOCKED + ) + UPDATE ${KERNL_SCHEMA_NAME}.scheduled_wakeups AS sw + SET claimed_at_s = $1, + updated_at = $3 + FROM due + WHERE sw.id = due.id + RETURNING sw.*; + `, + [nowS, limit, nowMsNum], + ); + + return result.rows.map((row) => + ScheduledWakeupCodec.decode( + ScheduledWakeupRecordSchema.parse(row), + ), ); } } From 778f55485696669990310a374040cac9ded83d52 Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Mon, 8 Dec 2025 11:20:07 -0700 Subject: [PATCH 04/25] Wakeup Sys Tool --- packages/kernl/src/tool/sys/index.ts | 2 + packages/kernl/src/tool/sys/wakeup.ts | 164 ++++++++++++++++++++++++++ 2 files changed, 166 insertions(+) create mode 100644 packages/kernl/src/tool/sys/wakeup.ts diff --git a/packages/kernl/src/tool/sys/index.ts b/packages/kernl/src/tool/sys/index.ts index fc1a7db0..a7e71751 100644 --- a/packages/kernl/src/tool/sys/index.ts +++ b/packages/kernl/src/tool/sys/index.ts @@ -8,3 +8,5 @@ export { memory } from "./memory"; +// TODO: This should honestly be called sleep. But semantics. +export { wakeup } from "./wakeup"; \ No newline at end of file diff --git a/packages/kernl/src/tool/sys/wakeup.ts b/packages/kernl/src/tool/sys/wakeup.ts new file mode 100644 index 00000000..2c30577b --- /dev/null +++ b/packages/kernl/src/tool/sys/wakeup.ts @@ -0,0 +1,164 @@ +/** + * Wakeup system toolkit. + * /packages/kernl/src/tool/sys/wakeup.ts + * + * Provides a tool for agents to schedule a wakeup (sleep/wait) for the current thread. + * The tool: + * - Creates a scheduled wakeup for the current thread + * - Returns INTERRUPTIBLE state so the thread can be stopped/checkpointed + * + * Enabled via a future `wakeup: true`-style config on the agent (parallel to memory). + */ + +import assert from "assert"; +import { z } from "zod"; + +import { tool } from "../tool"; +import { Toolkit } from "../toolkit"; + +/** + * Parameters for the wait tool: + * - delay_s: relative delay in seconds + * - run_at_s: absolute epoch seconds when the wakeup should fire + * + * Exactly one of delay_s or run_at_s must be provided. + */ +const WaitParamsSchema = z + .object({ + delay_s: z + .number() + .int() + .nonnegative() + .optional() + .describe( + "How many seconds from now to wait before resuming this thread.", + ), + + run_at_s: z + .number() + .int() + .nonnegative() + .optional() + .describe( + "Exact epoch seconds when this thread should be resumed. If provided, delay_s is ignored.", + ), + + reason: z + .string() + .max(512) + .optional() + .describe("Optional human-readable reason for the wakeup."), + }) + .refine( + (v) => (v.delay_s ?? null) !== null || (v.run_at_s ?? null) !== null, + { + message: "Either delay_s or run_at_s must be provided", + path: ["delay_s"], + }, + ) + .refine( + (v) => !((v.delay_s ?? null) !== null && (v.run_at_s ?? null) !== null), + { + message: "Provide either delay_s or run_at_s, but not both", + path: ["run_at_s"], + }, + ); + +/** + * Wait tool: + * - Schedules a wakeup for the current thread + * - Uses the approval path to return INTERRUPTIBLE, so the thread sleeps + * + * NOTE: The actual scheduling side-effect is performed inside `requiresApproval`, + * so that the tool returns state=INTERRUPTIBLE *without* ever running execute(). + */ +const wait = tool({ + id: "wait_until", + description: + "Pause this agent until a future time by scheduling a wakeup for the current thread " + + "and suspending execution. Use this to 'sleep' and resume later.", + + // This indicates the tool is conceptually async (long-running / external). + mode: "async" as const, + + parameters: WaitParamsSchema, + + /** + * Side-effect: schedule a wakeup and then return true so the tool call becomes INTERRUPTIBLE. + * + * We intentionally do the scheduling here (in requiresApproval) rather than in execute: + * - If this returns true and there is no approval recorded, the FunctionTool + * returns `{ state: INTERRUPTIBLE, result: undefined }` and *does not* call execute(). + * - That is exactly what we want for "sleep": schedule + interrupt. + * + * Expected environment contract (to wire WakeupStore): + * - ctx.agent MUST exist. + * - ctx.agent.wakeups.scheduleForCurrentThread({ run_at_s, reason }) MUST exist and: + * - create a ScheduledWakeup row tied to the current thread id + * - use your `WakeupStore` + `ScheduledWakeupRecord` schema + */ + requiresApproval: async (ctx, params) => { + assert(ctx.agent, "ctx.agent required for wakeup tools"); + + const { delay_s, run_at_s, reason } = params as z.infer< + typeof WaitParamsSchema + >; + + const now_s = Math.floor(Date.now() / 1000); + const targetRunAt = + run_at_s ?? (delay_s !== undefined ? now_s + delay_s : now_s); + + // We avoid over-constraining the Agent type here by treating it as `any`, + // and only requiring a small, well-defined surface: + // + // agent.wakeups.scheduleForCurrentThread({ run_at_s, reason? }) + // + // You can implement this on Agent however you like, backed by WakeupStore. + const agent: any = ctx.agent; + + if (!agent.wakeups || typeof agent.wakeups.scheduleForCurrentThread !== "function") { + throw new Error( + "Wakeup scheduler not configured on agent. " + + "Expected ctx.agent.wakeups.scheduleForCurrentThread(...) to be available.", + ); + } + + await agent.wakeups.scheduleForCurrentThread({ + run_at_s: targetRunAt, + reason: reason ?? null, + }); + + // Returning true here causes the FunctionTool to return INTERRUPTIBLE + // (because there is no approval yet), which will stop/sleep the thread. + return true; + }, + + /** + * execute() is not expected to run before the wakeup. + * + * If you *later* implement a resume-path that "approves" this tool call + * (e.g. via your scheduler/poller marking the tool call approved), this + * execute() would run on resume. For now, it simply returns a small + * acknowledgment. + */ + execute: async () => { + return { + scheduled: true, + message: + "Wakeup scheduled; thread will resume when the wakeup is processed.", + }; + }, +}); + +// --- Toolkit --- + +/** + * Wakeup system toolkit. + * + * Provides the wait_until tool for scheduling wakeups of the current thread. + */ +export const wakeup = new Toolkit({ + id: "sys.wakeup", + description: "Tools for scheduling and managing agent wakeups.", + tools: [wait], +}); From 4949a3bfb509697b7d5f0e3bac8028d3d975f03e Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Mon, 8 Dec 2025 11:47:11 -0700 Subject: [PATCH 05/25] Routes to files --- packages/kernl/src/agent.ts | 4 ++++ packages/kernl/src/agent/index.ts | 1 + packages/kernl/src/agent/types.ts | 4 ++++ packages/kernl/src/kernl/index.ts | 3 +++ packages/kernl/src/kernl/kernl.ts | 3 +++ packages/kernl/src/kernl/types.ts | 3 +++ 6 files changed, 18 insertions(+) diff --git a/packages/kernl/src/agent.ts b/packages/kernl/src/agent.ts index 3aebec08..3fdb66d1 100644 --- a/packages/kernl/src/agent.ts +++ b/packages/kernl/src/agent.ts @@ -1,3 +1,7 @@ +/** + * /packages/kernl/src/agent.ts + */ + import { message, LanguageModel, diff --git a/packages/kernl/src/agent/index.ts b/packages/kernl/src/agent/index.ts index e69de29b..a5de3b36 100644 --- a/packages/kernl/src/agent/index.ts +++ b/packages/kernl/src/agent/index.ts @@ -0,0 +1 @@ +// TODO: Ask Andrew, Not sure if this should be blank \ No newline at end of file diff --git a/packages/kernl/src/agent/types.ts b/packages/kernl/src/agent/types.ts index fb392939..8221a73e 100644 --- a/packages/kernl/src/agent/types.ts +++ b/packages/kernl/src/agent/types.ts @@ -1,3 +1,7 @@ +/** + * /packages/kernl/src/agent/types.ts + */ + import { type ZodType } from "zod"; import { diff --git a/packages/kernl/src/kernl/index.ts b/packages/kernl/src/kernl/index.ts index cd050863..8f2bfa1e 100644 --- a/packages/kernl/src/kernl/index.ts +++ b/packages/kernl/src/kernl/index.ts @@ -1,3 +1,6 @@ +/** + * /packages/kernl/src/kernl/index.ts + */ export { Kernl } from "./kernl"; export type { KernlOptions, diff --git a/packages/kernl/src/kernl/kernl.ts b/packages/kernl/src/kernl/kernl.ts index cc00600f..14ae3b45 100644 --- a/packages/kernl/src/kernl/kernl.ts +++ b/packages/kernl/src/kernl/kernl.ts @@ -1,3 +1,6 @@ +/** + * /packages/kernl/src/kernl/kernl.ts + */ import type { LanguageModel } from "@kernl-sdk/protocol"; import { resolveEmbeddingModel } from "@kernl-sdk/retrieval"; diff --git a/packages/kernl/src/kernl/types.ts b/packages/kernl/src/kernl/types.ts index 44972cc3..07514b8a 100644 --- a/packages/kernl/src/kernl/types.ts +++ b/packages/kernl/src/kernl/types.ts @@ -1,3 +1,6 @@ +/** + * /packages/kernl/src/kernl/types.ts + */ import { LanguageModel, EmbeddingModel } from "@kernl-sdk/protocol"; import { SearchIndex } from "@kernl-sdk/retrieval"; From 136a2ee56d6018bd84a316424ee6936bdfd2d665 Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Mon, 8 Dec 2025 12:08:19 -0700 Subject: [PATCH 06/25] [Builds Locally] Add in-memory wakeup store to Kernl storage Introduces the WakeupStore interface to KernlStorage and implements an in-memory version (InMemoryWakeupStore) for managing scheduled wakeups. Updates the in-memory storage class to include wakeup management, supporting creation, retrieval, update, deletion, and claiming of due wakeups. --- packages/kernl/src/storage/base.ts | 10 ++- packages/kernl/src/storage/in-memory.ts | 103 ++++++++++++++++++++++++ packages/kernl/src/storage/thread.ts | 1 + 3 files changed, 113 insertions(+), 1 deletion(-) diff --git a/packages/kernl/src/storage/base.ts b/packages/kernl/src/storage/base.ts index cf42ad63..d5744725 100644 --- a/packages/kernl/src/storage/base.ts +++ b/packages/kernl/src/storage/base.ts @@ -1,10 +1,13 @@ /** * Core storage contracts. + * /packages/kernl/src/storage/base.ts */ import type { AgentRegistry, ModelRegistry } from "@/kernl/types"; import type { ThreadStore } from "./thread"; -import type { MemoryStore } from "@/memory/store"; +import type { MemoryStore } from "@/memory" +// import type { MemoryStore } from "@/memory/store" +import type { WakeupStore } from "@/wakeup"; /** * The main storage interface for Kernl. @@ -22,6 +25,11 @@ export interface KernlStorage { */ memories: MemoryStore; + /** + * Wakeup store - manages wakeup records for agents. + */ + wakeups: WakeupStore + // tasks: TaskStore; // traces: TraceStore; diff --git a/packages/kernl/src/storage/in-memory.ts b/packages/kernl/src/storage/in-memory.ts index 6cc0b3ae..e51b6b6b 100644 --- a/packages/kernl/src/storage/in-memory.ts +++ b/packages/kernl/src/storage/in-memory.ts @@ -1,5 +1,6 @@ /** * In-memory storage implementation for Kernl. + * /packages/kernl/src/storage/in-memory.ts * * Pure domain-level - no codecs, schemas, or DB records. * @@ -33,6 +34,13 @@ import type { MemoryListOptions, MemoryFilter, } from "@/memory"; +import type { + NewScheduledWakeup, + ScheduledWakeup, + ScheduledWakeupUpdate, + WakeupStore, +} from "@/wakeup"; + /** * In-memory storage implementation. @@ -40,10 +48,12 @@ import type { export class InMemoryStorage implements KernlStorage { threads: InMemoryThreadStore; memories: InMemoryMemoryStore; + wakeups: InMemoryWakeupStore; constructor() { this.threads = new InMemoryThreadStore(); this.memories = new InMemoryMemoryStore(); + this.wakeups = new InMemoryWakeupStore(); } bind(registries: { agents: AgentRegistry; models: ModelRegistry }): void { @@ -542,3 +552,96 @@ export class InMemoryMemoryStore implements MemoryStore { }); } } + + +/** + * In-memory wakeup store implementation + */ +export class InMemoryWakeupStore implements WakeupStore { + private wakeups = new Map(); + + async create(input: NewScheduledWakeup): Promise { + const now = Date.now(); + + const record: ScheduledWakeup = { + id: input.id, + threadId: input.threadId, + runAt: input.runAt, + reason: input.reason ?? null, + woken: false, + claimedAt: null, + createdAt: now, + updatedAt: now, + error: null, + }; + + this.wakeups.set(record.id, record); + return record; + } + + async get(id: string): Promise { + return this.wakeups.get(id) ?? null; + } + + async update( + id: string, + patch: ScheduledWakeupUpdate, + ): Promise { + const existing = this.wakeups.get(id); + if (!existing) { + throw new Error(`Wakeup with id ${id} not found`); + } + + const updated: ScheduledWakeup = { + ...existing, + runAt: patch.runAt ?? existing.runAt, + reason: patch.reason ?? existing.reason, + woken: patch.woken ?? existing.woken, + claimedAt: + patch.claimedAt !== undefined ? patch.claimedAt : existing.claimedAt, + error: patch.error ?? existing.error, + updatedAt: patch.updatedAt ?? Date.now(), + }; + + this.wakeups.set(id, updated); + return updated; + } + + async delete(id: string): Promise { + this.wakeups.delete(id); + } + + async claimDue( + nowMs: number | bigint, + limit: number, + ): Promise { + const now = typeof nowMs === "bigint" ? Number(nowMs) : nowMs; + + const candidates: ScheduledWakeup[] = []; + + for (const w of this.wakeups.values()) { + if (!w.woken && w.claimedAt == null && w.runAt <= now) { + candidates.push(w); + } + } + + candidates.sort((a, b) => a.runAt - b.runAt); + + const selected = candidates.slice(0, limit); + const updatedSelected: ScheduledWakeup[] = []; + + const claimTime = Date.now(); + + for (const w of selected) { + const updated: ScheduledWakeup = { + ...w, + claimedAt: claimTime, + updatedAt: claimTime, + }; + this.wakeups.set(updated.id, updated); + updatedSelected.push(updated); + } + + return updatedSelected; + } +} diff --git a/packages/kernl/src/storage/thread.ts b/packages/kernl/src/storage/thread.ts index 566738db..fa94b77c 100644 --- a/packages/kernl/src/storage/thread.ts +++ b/packages/kernl/src/storage/thread.ts @@ -1,5 +1,6 @@ /** * Thread storage contracts. + * /packages/kernl/src/storage/thread.ts */ import type { Thread } from "@/thread"; From 62deb7fb5cdc43c3c7ce8cf6710ca27f0275676f Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Mon, 8 Dec 2025 12:53:51 -0700 Subject: [PATCH 07/25] [Builds Locally] Add wakeup system tool and integrate with agent Introduces a new 'wakeup' system tool for scheduling thread wakeups, exposes it via the toolkit system, and ensures it is initialized and bound in the Agent class. Updates the wait tool's parameter schema and documentation for clarity, and exports the wakeup toolkit alongside memory in the tool index. --- packages/kernl/src/agent.ts | 11 +- packages/kernl/src/storage/base.ts | 3 +- packages/kernl/src/tool/index.ts | 2 +- packages/kernl/src/tool/sys/wakeup.ts | 259 ++++++++++++++++++++------ 4 files changed, 214 insertions(+), 61 deletions(-) diff --git a/packages/kernl/src/agent.ts b/packages/kernl/src/agent.ts index 3fdb66d1..02872e36 100644 --- a/packages/kernl/src/agent.ts +++ b/packages/kernl/src/agent.ts @@ -19,7 +19,7 @@ import type { RThreadUpdateParams, } from "@/api/resources/threads/types"; import type { Context, UnknownContext } from "./context"; -import { Tool, memory } from "./tool"; +import { Tool, memory, wakeup } from "./tool"; import { BaseToolkit } from "./tool/toolkit"; import { InputGuardrail, @@ -116,14 +116,23 @@ export class Agent< this.kernl = kernl; // initialize system toolkits + + // Memory System Tool if (this.memory.enabled) { // safety: system tools only rely on ctx.agent, not ctx.context const toolkit = memory as unknown as BaseToolkit; this.systools.push(toolkit); toolkit.bind(this); } + // Wakeup System Tool + { + const wakeupToolKit = wakeup as unknown as BaseToolkit; + this.systools.push(wakeupToolKit); + wakeupToolKit.bind(this); + } } + /** * Blocking execution - spawns or resumes thread and waits for completion * diff --git a/packages/kernl/src/storage/base.ts b/packages/kernl/src/storage/base.ts index d5744725..90e1736a 100644 --- a/packages/kernl/src/storage/base.ts +++ b/packages/kernl/src/storage/base.ts @@ -6,7 +6,6 @@ import type { AgentRegistry, ModelRegistry } from "@/kernl/types"; import type { ThreadStore } from "./thread"; import type { MemoryStore } from "@/memory" -// import type { MemoryStore } from "@/memory/store" import type { WakeupStore } from "@/wakeup"; /** @@ -28,7 +27,7 @@ export interface KernlStorage { /** * Wakeup store - manages wakeup records for agents. */ - wakeups: WakeupStore + wakeups: WakeupStore; // tasks: TaskStore; // traces: TraceStore; diff --git a/packages/kernl/src/tool/index.ts b/packages/kernl/src/tool/index.ts index 3c5488b0..abf25ef3 100644 --- a/packages/kernl/src/tool/index.ts +++ b/packages/kernl/src/tool/index.ts @@ -14,4 +14,4 @@ export type { } from "./types"; // --- system toolkits --- -export { memory } from "./sys"; +export { memory, wakeup } from "./sys"; diff --git a/packages/kernl/src/tool/sys/wakeup.ts b/packages/kernl/src/tool/sys/wakeup.ts index 2c30577b..d5ef934d 100644 --- a/packages/kernl/src/tool/sys/wakeup.ts +++ b/packages/kernl/src/tool/sys/wakeup.ts @@ -17,11 +17,11 @@ import { tool } from "../tool"; import { Toolkit } from "../toolkit"; /** - * Parameters for the wait tool: - * - delay_s: relative delay in seconds - * - run_at_s: absolute epoch seconds when the wakeup should fire + * Parameters for the wait tool. * - * Exactly one of delay_s or run_at_s must be provided. + * At least one of: + * - delay_s: number of seconds from now + * - run_at_s: absolute epoch seconds */ const WaitParamsSchema = z .object({ @@ -33,16 +33,12 @@ const WaitParamsSchema = z .describe( "How many seconds from now to wait before resuming this thread.", ), - run_at_s: z .number() .int() .nonnegative() .optional() - .describe( - "Exact epoch seconds when this thread should be resumed. If provided, delay_s is ignored.", - ), - + .describe("Exact epoch seconds when this thread should be resumed."), reason: z .string() .max(512) @@ -55,48 +51,33 @@ const WaitParamsSchema = z message: "Either delay_s or run_at_s must be provided", path: ["delay_s"], }, - ) - .refine( - (v) => !((v.delay_s ?? null) !== null && (v.run_at_s ?? null) !== null), - { - message: "Provide either delay_s or run_at_s, but not both", - path: ["run_at_s"], - }, ); /** - * Wait tool: - * - Schedules a wakeup for the current thread - * - Uses the approval path to return INTERRUPTIBLE, so the thread sleeps + * wait_until + * + * Schedules a wakeup for the current thread and then causes the tool call + * to be INTERRUPTIBLE (via requiresApproval). + * + * Contract: + * - ctx.agent MUST exist. + * - ctx.agent.wakeups.scheduleForCurrentThread({ run_at_s, reason }) MUST exist and: + * - create a ScheduledWakeup row tied to the current thread id + * - use your WakeupStore + ScheduledWakeupRecord schema * - * NOTE: The actual scheduling side-effect is performed inside `requiresApproval`, - * so that the tool returns state=INTERRUPTIBLE *without* ever running execute(). + * Tool engine behavior: + * - For async tools with requiresApproval: + * - if requiresApproval returns true, the engine treats the call as + * INTERRUPTIBLE and does not immediately run execute(). + * - that is exactly what we want for "sleep": schedule + interrupt. */ const wait = tool({ id: "wait_until", description: - "Pause this agent until a future time by scheduling a wakeup for the current thread " + - "and suspending execution. Use this to 'sleep' and resume later.", - - // This indicates the tool is conceptually async (long-running / external). + "Pause this agent until a future time by scheduling a wakeup for the current thread.", mode: "async" as const, - parameters: WaitParamsSchema, - /** - * Side-effect: schedule a wakeup and then return true so the tool call becomes INTERRUPTIBLE. - * - * We intentionally do the scheduling here (in requiresApproval) rather than in execute: - * - If this returns true and there is no approval recorded, the FunctionTool - * returns `{ state: INTERRUPTIBLE, result: undefined }` and *does not* call execute(). - * - That is exactly what we want for "sleep": schedule + interrupt. - * - * Expected environment contract (to wire WakeupStore): - * - ctx.agent MUST exist. - * - ctx.agent.wakeups.scheduleForCurrentThread({ run_at_s, reason }) MUST exist and: - * - create a ScheduledWakeup row tied to the current thread id - * - use your `WakeupStore` + `ScheduledWakeupRecord` schema - */ requiresApproval: async (ctx, params) => { assert(ctx.agent, "ctx.agent required for wakeup tools"); @@ -105,41 +86,39 @@ const wait = tool({ >; const now_s = Math.floor(Date.now() / 1000); - const targetRunAt = + const targetRunAt_s = run_at_s ?? (delay_s !== undefined ? now_s + delay_s : now_s); - // We avoid over-constraining the Agent type here by treating it as `any`, - // and only requiring a small, well-defined surface: - // - // agent.wakeups.scheduleForCurrentThread({ run_at_s, reason? }) - // - // You can implement this on Agent however you like, backed by WakeupStore. const agent: any = ctx.agent; - if (!agent.wakeups || typeof agent.wakeups.scheduleForCurrentThread !== "function") { + if ( + !agent.wakeups || + typeof agent.wakeups.scheduleForCurrentThread !== "function" + ) { throw new Error( - "Wakeup scheduler not configured on agent. " + - "Expected ctx.agent.wakeups.scheduleForCurrentThread(...) to be available.", + "Expected ctx.agent.wakeups.scheduleForCurrentThread(...) to be available.", ); } + // Delegate to the agent-level wakeup service so we don't depend on + // ctx.kernl or ctx.thread here. await agent.wakeups.scheduleForCurrentThread({ - run_at_s: targetRunAt, + run_at_s: targetRunAt_s, reason: reason ?? null, }); - // Returning true here causes the FunctionTool to return INTERRUPTIBLE - // (because there is no approval yet), which will stop/sleep the thread. + // Returning true tells the tool engine: + // - this call requires approval + // - mark the tool call as INTERRUPTIBLE and *don't* run execute() yet return true; }, /** * execute() is not expected to run before the wakeup. * - * If you *later* implement a resume-path that "approves" this tool call - * (e.g. via your scheduler/poller marking the tool call approved), this - * execute() would run on resume. For now, it simply returns a small - * acknowledgment. + * If you later implement a resume path that "approves" this tool call + * (e.g. scheduler/poller marks it approved), execute() will run on resume. + * For now, it just returns a small acknowledgement. */ execute: async () => { return { @@ -162,3 +141,169 @@ export const wakeup = new Toolkit({ description: "Tools for scheduling and managing agent wakeups.", tools: [wait], }); + + +// /** +// * Wakeup system toolkit. +// * /packages/kernl/src/tool/sys/wakeup.ts +// * +// * Provides a tool for agents to schedule a wakeup (sleep/wait) for the current thread. +// * The tool: +// * - Creates a scheduled wakeup for the current thread +// * - Returns INTERRUPTIBLE state so the thread can be stopped/checkpointed +// * +// * Enabled via a future `wakeup: true`-style config on the agent (parallel to memory). +// */ + +// import assert from "assert"; +// import { z } from "zod"; + +// import { tool } from "../tool"; +// import { Toolkit } from "../toolkit"; + +// /** +// * Parameters for the wait tool: +// * - delay_s: relative delay in seconds +// * - run_at_s: absolute epoch seconds when the wakeup should fire +// * +// * Exactly one of delay_s or run_at_s must be provided. +// */ +// const WaitParamsSchema = z +// .object({ +// delay_s: z +// .number() +// .int() +// .nonnegative() +// .optional() +// .describe( +// "How many seconds from now to wait before resuming this thread.", +// ), + +// run_at_s: z +// .number() +// .int() +// .nonnegative() +// .optional() +// .describe( +// "Exact epoch seconds when this thread should be resumed. If provided, delay_s is ignored.", +// ), + +// reason: z +// .string() +// .max(512) +// .optional() +// .describe("Optional human-readable reason for the wakeup."), +// }) +// .refine( +// (v) => (v.delay_s ?? null) !== null || (v.run_at_s ?? null) !== null, +// { +// message: "Either delay_s or run_at_s must be provided", +// path: ["delay_s"], +// }, +// ) +// .refine( +// (v) => !((v.delay_s ?? null) !== null && (v.run_at_s ?? null) !== null), +// { +// message: "Provide either delay_s or run_at_s, but not both", +// path: ["run_at_s"], +// }, +// ); + +// /** +// * Wait tool: +// * - Schedules a wakeup for the current thread +// * - Uses the approval path to return INTERRUPTIBLE, so the thread sleeps +// * +// * NOTE: The actual scheduling side-effect is performed inside `requiresApproval`, +// * so that the tool returns state=INTERRUPTIBLE *without* ever running execute(). +// */ +// const wait = tool({ +// id: "wait_until", +// description: +// "Pause this agent until a future time by scheduling a wakeup for the current thread " + +// "and suspending execution. Use this to 'sleep' and resume later.", + +// // This indicates the tool is conceptually async (long-running / external). +// mode: "async" as const, + +// parameters: WaitParamsSchema, + +// /** +// * Side-effect: schedule a wakeup and then return true so the tool call becomes INTERRUPTIBLE. +// * +// * We intentionally do the scheduling here (in requiresApproval) rather than in execute: +// * - If this returns true and there is no approval recorded, the FunctionTool +// * returns `{ state: INTERRUPTIBLE, result: undefined }` and *does not* call execute(). +// * - That is exactly what we want for "sleep": schedule + interrupt. +// * +// * Expected environment contract (to wire WakeupStore): +// * - ctx.agent MUST exist. +// * - ctx.agent.wakeups.scheduleForCurrentThread({ run_at_s, reason }) MUST exist and: +// * - create a ScheduledWakeup row tied to the current thread id +// * - use your `WakeupStore` + `ScheduledWakeupRecord` schema +// */ +// requiresApproval: async (ctx, params) => { +// assert(ctx.agent, "ctx.agent required for wakeup tools"); + +// const { delay_s, run_at_s, reason } = params as z.infer< +// typeof WaitParamsSchema +// >; + +// const now_s = Math.floor(Date.now() / 1000); +// const targetRunAt = +// run_at_s ?? (delay_s !== undefined ? now_s + delay_s : now_s); + +// // We avoid over-constraining the Agent type here by treating it as `any`, +// // and only requiring a small, well-defined surface: +// // +// // agent.wakeups.scheduleForCurrentThread({ run_at_s, reason? }) +// // +// // You can implement this on Agent however you like, backed by WakeupStore. +// const agent: any = ctx.agent; + +// if (!agent.wakeups || typeof agent.wakeups.scheduleForCurrentThread !== "function") { +// throw new Error( +// "Wakeup scheduler not configured on agent. " + +// "Expected ctx.agent.wakeups.scheduleForCurrentThread(...) to be available.", +// ); +// } + +// await agent.wakeups.scheduleForCurrentThread({ +// run_at_s: targetRunAt, +// reason: reason ?? null, +// }); + +// // Returning true here causes the FunctionTool to return INTERRUPTIBLE +// // (because there is no approval yet), which will stop/sleep the thread. +// return true; +// }, + +// /** +// * execute() is not expected to run before the wakeup. +// * +// * If you *later* implement a resume-path that "approves" this tool call +// * (e.g. via your scheduler/poller marking the tool call approved), this +// * execute() would run on resume. For now, it simply returns a small +// * acknowledgment. +// */ +// execute: async () => { +// return { +// scheduled: true, +// message: +// "Wakeup scheduled; thread will resume when the wakeup is processed.", +// }; +// }, +// }); + +// // --- Toolkit --- + +// /** +// * Wakeup system toolkit. +// * +// * Provides the wait_until tool for scheduling wakeups of the current thread. +// */ +// export const wakeup = new Toolkit({ +// id: "sys.wakeup", +// description: "Tools for scheduling and managing agent wakeups.", +// tools: [wait], +// }); From fde226951d208e9ece05645aada57c91ffda4841 Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Mon, 8 Dec 2025 12:55:40 -0700 Subject: [PATCH 08/25] [Builds Locally] Refactor wakeup tool to require explicit thread_id Updated the wait_until tool to require the runtime to pass an explicit thread_id parameter, rather than relying on agent-level scheduling for the current thread. The tool now schedules wakeups directly via the storage-backed wakeup store, improving clarity and decoupling from agent internals. Documentation and parameter schema were updated accordingly. --- packages/kernl/src/tool/sys/wakeup.ts | 263 +++++--------------------- 1 file changed, 45 insertions(+), 218 deletions(-) diff --git a/packages/kernl/src/tool/sys/wakeup.ts b/packages/kernl/src/tool/sys/wakeup.ts index d5ef934d..4e469f5b 100644 --- a/packages/kernl/src/tool/sys/wakeup.ts +++ b/packages/kernl/src/tool/sys/wakeup.ts @@ -3,11 +3,10 @@ * /packages/kernl/src/tool/sys/wakeup.ts * * Provides a tool for agents to schedule a wakeup (sleep/wait) for the current thread. - * The tool: - * - Creates a scheduled wakeup for the current thread - * - Returns INTERRUPTIBLE state so the thread can be stopped/checkpointed * - * Enabled via a future `wakeup: true`-style config on the agent (parallel to memory). + * The thread runtime is responsible for: + * - Passing the current thread_id into this tool's parameters. + * - Using the async/INTERRUPTIBLE tool state to checkpoint/save and stop the thread. */ import assert from "assert"; @@ -19,12 +18,18 @@ import { Toolkit } from "../toolkit"; /** * Parameters for the wait tool. * - * At least one of: - * - delay_s: number of seconds from now - * - run_at_s: absolute epoch seconds + * The runtime must pass: + * - thread_id: the id of the thread to resume later + * And either: + * - delay_s: number of seconds from now, OR + * - run_at_s: absolute epoch seconds when the thread should resume */ const WaitParamsSchema = z .object({ + thread_id: z + .string() + .min(1) + .describe("The id of the current thread to resume later."), delay_s: z .number() .int() @@ -56,254 +61,76 @@ const WaitParamsSchema = z /** * wait_until * - * Schedules a wakeup for the current thread and then causes the tool call - * to be INTERRUPTIBLE (via requiresApproval). - * - * Contract: - * - ctx.agent MUST exist. - * - ctx.agent.wakeups.scheduleForCurrentThread({ run_at_s, reason }) MUST exist and: - * - create a ScheduledWakeup row tied to the current thread id - * - use your WakeupStore + ScheduledWakeupRecord schema - * - * Tool engine behavior: - * - For async tools with requiresApproval: - * - if requiresApproval returns true, the engine treats the call as - * INTERRUPTIBLE and does not immediately run execute(). - * - that is exactly what we want for "sleep": schedule + interrupt. + * Schedules a wakeup for the given thread_id and then causes the tool call + * to be INTERRUPTIBLE (via requiresApproval). The actual checkpoint/save/stop + * behaviour is handled by the existing async tool + thread runtime. */ const wait = tool({ id: "wait_until", description: - "Pause this agent until a future time by scheduling a wakeup for the current thread.", + "Pause this agent until a future time by scheduling a wakeup for the given thread.", mode: "async" as const, parameters: WaitParamsSchema, + /** + * We do the scheduling here and return true so the tool call becomes + * INTERRUPTIBLE and execute() is not run until some future approval/resume. + */ requiresApproval: async (ctx, params) => { - assert(ctx.agent, "ctx.agent required for wakeup tools"); + assert(ctx.agent, "ctx.agent is required for wakeup tools"); - const { delay_s, run_at_s, reason } = params as z.infer< + const { thread_id, delay_s, run_at_s, reason } = params as z.infer< typeof WaitParamsSchema >; - const now_s = Math.floor(Date.now() / 1000); - const targetRunAt_s = - run_at_s ?? (delay_s !== undefined ? now_s + delay_s : now_s); - const agent: any = ctx.agent; - if ( - !agent.wakeups || - typeof agent.wakeups.scheduleForCurrentThread !== "function" - ) { - throw new Error( - "Expected ctx.agent.wakeups.scheduleForCurrentThread(...) to be available.", - ); + if (!agent.kernl || !agent.kernl.storage) { + throw new Error("Agent is not bound to Kernl storage."); + } + + const wakeupStore: any = agent.kernl.storage.wakeups; + if (!wakeupStore || typeof wakeupStore.create !== "function") { + throw new Error("Wakeup store is not configured on Kernl storage."); } - // Delegate to the agent-level wakeup service so we don't depend on - // ctx.kernl or ctx.thread here. - await agent.wakeups.scheduleForCurrentThread({ + const now_s = Math.floor(Date.now() / 1000); + const targetRunAt_s = + run_at_s ?? (delay_s !== undefined ? now_s + delay_s : now_s); + + // Delegate to the storage-backed wakeup store. + // We keep the shape generic here and rely on the store implementation + // to map fields into its internal schema. + await wakeupStore.create({ + thread_id, run_at_s: targetRunAt_s, reason: reason ?? null, }); - // Returning true tells the tool engine: - // - this call requires approval - // - mark the tool call as INTERRUPTIBLE and *don't* run execute() yet + // Returning true tells the tool engine this call requires approval, + // so it will mark the tool call as INTERRUPTIBLE and *not* run execute() + // until a scheduler/poller resumes/approves it. return true; }, /** - * execute() is not expected to run before the wakeup. - * - * If you later implement a resume path that "approves" this tool call - * (e.g. scheduler/poller marks it approved), execute() will run on resume. - * For now, it just returns a small acknowledgement. + * execute() is only expected to run if/when the tool call is explicitly + * approved/resumed by your scheduler. For now it just returns a simple + * acknowledgement. */ execute: async () => { return { scheduled: true, message: - "Wakeup scheduled; thread will resume when the wakeup is processed.", + "Wakeup scheduled; the thread will resume when the wakeup is processed.", }; }, }); // --- Toolkit --- -/** - * Wakeup system toolkit. - * - * Provides the wait_until tool for scheduling wakeups of the current thread. - */ export const wakeup = new Toolkit({ id: "sys.wakeup", description: "Tools for scheduling and managing agent wakeups.", tools: [wait], }); - - -// /** -// * Wakeup system toolkit. -// * /packages/kernl/src/tool/sys/wakeup.ts -// * -// * Provides a tool for agents to schedule a wakeup (sleep/wait) for the current thread. -// * The tool: -// * - Creates a scheduled wakeup for the current thread -// * - Returns INTERRUPTIBLE state so the thread can be stopped/checkpointed -// * -// * Enabled via a future `wakeup: true`-style config on the agent (parallel to memory). -// */ - -// import assert from "assert"; -// import { z } from "zod"; - -// import { tool } from "../tool"; -// import { Toolkit } from "../toolkit"; - -// /** -// * Parameters for the wait tool: -// * - delay_s: relative delay in seconds -// * - run_at_s: absolute epoch seconds when the wakeup should fire -// * -// * Exactly one of delay_s or run_at_s must be provided. -// */ -// const WaitParamsSchema = z -// .object({ -// delay_s: z -// .number() -// .int() -// .nonnegative() -// .optional() -// .describe( -// "How many seconds from now to wait before resuming this thread.", -// ), - -// run_at_s: z -// .number() -// .int() -// .nonnegative() -// .optional() -// .describe( -// "Exact epoch seconds when this thread should be resumed. If provided, delay_s is ignored.", -// ), - -// reason: z -// .string() -// .max(512) -// .optional() -// .describe("Optional human-readable reason for the wakeup."), -// }) -// .refine( -// (v) => (v.delay_s ?? null) !== null || (v.run_at_s ?? null) !== null, -// { -// message: "Either delay_s or run_at_s must be provided", -// path: ["delay_s"], -// }, -// ) -// .refine( -// (v) => !((v.delay_s ?? null) !== null && (v.run_at_s ?? null) !== null), -// { -// message: "Provide either delay_s or run_at_s, but not both", -// path: ["run_at_s"], -// }, -// ); - -// /** -// * Wait tool: -// * - Schedules a wakeup for the current thread -// * - Uses the approval path to return INTERRUPTIBLE, so the thread sleeps -// * -// * NOTE: The actual scheduling side-effect is performed inside `requiresApproval`, -// * so that the tool returns state=INTERRUPTIBLE *without* ever running execute(). -// */ -// const wait = tool({ -// id: "wait_until", -// description: -// "Pause this agent until a future time by scheduling a wakeup for the current thread " + -// "and suspending execution. Use this to 'sleep' and resume later.", - -// // This indicates the tool is conceptually async (long-running / external). -// mode: "async" as const, - -// parameters: WaitParamsSchema, - -// /** -// * Side-effect: schedule a wakeup and then return true so the tool call becomes INTERRUPTIBLE. -// * -// * We intentionally do the scheduling here (in requiresApproval) rather than in execute: -// * - If this returns true and there is no approval recorded, the FunctionTool -// * returns `{ state: INTERRUPTIBLE, result: undefined }` and *does not* call execute(). -// * - That is exactly what we want for "sleep": schedule + interrupt. -// * -// * Expected environment contract (to wire WakeupStore): -// * - ctx.agent MUST exist. -// * - ctx.agent.wakeups.scheduleForCurrentThread({ run_at_s, reason }) MUST exist and: -// * - create a ScheduledWakeup row tied to the current thread id -// * - use your `WakeupStore` + `ScheduledWakeupRecord` schema -// */ -// requiresApproval: async (ctx, params) => { -// assert(ctx.agent, "ctx.agent required for wakeup tools"); - -// const { delay_s, run_at_s, reason } = params as z.infer< -// typeof WaitParamsSchema -// >; - -// const now_s = Math.floor(Date.now() / 1000); -// const targetRunAt = -// run_at_s ?? (delay_s !== undefined ? now_s + delay_s : now_s); - -// // We avoid over-constraining the Agent type here by treating it as `any`, -// // and only requiring a small, well-defined surface: -// // -// // agent.wakeups.scheduleForCurrentThread({ run_at_s, reason? }) -// // -// // You can implement this on Agent however you like, backed by WakeupStore. -// const agent: any = ctx.agent; - -// if (!agent.wakeups || typeof agent.wakeups.scheduleForCurrentThread !== "function") { -// throw new Error( -// "Wakeup scheduler not configured on agent. " + -// "Expected ctx.agent.wakeups.scheduleForCurrentThread(...) to be available.", -// ); -// } - -// await agent.wakeups.scheduleForCurrentThread({ -// run_at_s: targetRunAt, -// reason: reason ?? null, -// }); - -// // Returning true here causes the FunctionTool to return INTERRUPTIBLE -// // (because there is no approval yet), which will stop/sleep the thread. -// return true; -// }, - -// /** -// * execute() is not expected to run before the wakeup. -// * -// * If you *later* implement a resume-path that "approves" this tool call -// * (e.g. via your scheduler/poller marking the tool call approved), this -// * execute() would run on resume. For now, it simply returns a small -// * acknowledgment. -// */ -// execute: async () => { -// return { -// scheduled: true, -// message: -// "Wakeup scheduled; thread will resume when the wakeup is processed.", -// }; -// }, -// }); - -// // --- Toolkit --- - -// /** -// * Wakeup system toolkit. -// * -// * Provides the wait_until tool for scheduling wakeups of the current thread. -// */ -// export const wakeup = new Toolkit({ -// id: "sys.wakeup", -// description: "Tools for scheduling and managing agent wakeups.", -// tools: [wait], -// }); From a9ba3026b80bbfdf5bb28509f0fdfee506d476c0 Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Mon, 8 Dec 2025 13:19:19 -0700 Subject: [PATCH 09/25] Update wakeup record creation to match domain schema Scheduled wakeup records now use generated IDs, camelCase field names, and store runAt in epoch milliseconds to align with the NewScheduledWakeup domain type. --- packages/kernl/src/tool/sys/wakeup.ts | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/packages/kernl/src/tool/sys/wakeup.ts b/packages/kernl/src/tool/sys/wakeup.ts index 4e469f5b..d36b6dec 100644 --- a/packages/kernl/src/tool/sys/wakeup.ts +++ b/packages/kernl/src/tool/sys/wakeup.ts @@ -12,6 +12,8 @@ import assert from "assert"; import { z } from "zod"; +import { randomID } from "@kernl-sdk/shared/lib"; + import { tool } from "../tool"; import { Toolkit } from "../toolkit"; @@ -97,13 +99,17 @@ const wait = tool({ const now_s = Math.floor(Date.now() / 1000); const targetRunAt_s = run_at_s ?? (delay_s !== undefined ? now_s + delay_s : now_s); + const targetRunAt_ms = targetRunAt_s * 1000; - // Delegate to the storage-backed wakeup store. - // We keep the shape generic here and rely on the store implementation - // to map fields into its internal schema. + // Create the scheduled wakeup record. + // Field names and units must match NewScheduledWakeup domain type: + // - id: unique wakeup ID (generated here) + // - threadId: camelCase + // - runAt: epoch milliseconds await wakeupStore.create({ - thread_id, - run_at_s: targetRunAt_s, + id: `wkp_${randomID()}`, + threadId: thread_id, + runAt: targetRunAt_ms, reason: reason ?? null, }); From 1c1c7981e0861fb66d9efce8001eaaf2722c40a4 Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Mon, 8 Dec 2025 13:30:47 -0700 Subject: [PATCH 10/25] [Builds Locally] Rename wakeup system tool to sleep Refactored the system toolkit previously named 'wakeup' to 'sleep' for improved semantic clarity. Updated all relevant imports, exports, and documentation to reflect the new naming. --- packages/kernl/src/agent.ts | 10 +++++----- packages/kernl/src/tool/index.ts | 2 +- packages/kernl/src/tool/sys/index.ts | 5 +---- packages/kernl/src/tool/sys/{wakeup.ts => sleep.ts} | 12 ++++++------ 4 files changed, 13 insertions(+), 16 deletions(-) rename packages/kernl/src/tool/sys/{wakeup.ts => sleep.ts} (93%) diff --git a/packages/kernl/src/agent.ts b/packages/kernl/src/agent.ts index 02872e36..bcbba5fc 100644 --- a/packages/kernl/src/agent.ts +++ b/packages/kernl/src/agent.ts @@ -19,7 +19,7 @@ import type { RThreadUpdateParams, } from "@/api/resources/threads/types"; import type { Context, UnknownContext } from "./context"; -import { Tool, memory, wakeup } from "./tool"; +import { Tool, memory, sleep } from "./tool"; import { BaseToolkit } from "./tool/toolkit"; import { InputGuardrail, @@ -124,11 +124,11 @@ export class Agent< this.systools.push(toolkit); toolkit.bind(this); } - // Wakeup System Tool + // Sleep System Tool { - const wakeupToolKit = wakeup as unknown as BaseToolkit; - this.systools.push(wakeupToolKit); - wakeupToolKit.bind(this); + const sleepToolkit = sleep as unknown as BaseToolkit; + this.systools.push(sleepToolkit); + sleepToolkit.bind(this); } } diff --git a/packages/kernl/src/tool/index.ts b/packages/kernl/src/tool/index.ts index abf25ef3..a2844fde 100644 --- a/packages/kernl/src/tool/index.ts +++ b/packages/kernl/src/tool/index.ts @@ -14,4 +14,4 @@ export type { } from "./types"; // --- system toolkits --- -export { memory, wakeup } from "./sys"; +export { memory, sleep } from "./sys"; diff --git a/packages/kernl/src/tool/sys/index.ts b/packages/kernl/src/tool/sys/index.ts index a7e71751..32421576 100644 --- a/packages/kernl/src/tool/sys/index.ts +++ b/packages/kernl/src/tool/sys/index.ts @@ -5,8 +5,5 @@ * These are internal toolkits that can be enabled via agent config flags. */ - - export { memory } from "./memory"; -// TODO: This should honestly be called sleep. But semantics. -export { wakeup } from "./wakeup"; \ No newline at end of file +export { sleep } from "./sleep"; \ No newline at end of file diff --git a/packages/kernl/src/tool/sys/wakeup.ts b/packages/kernl/src/tool/sys/sleep.ts similarity index 93% rename from packages/kernl/src/tool/sys/wakeup.ts rename to packages/kernl/src/tool/sys/sleep.ts index d36b6dec..fdf096d5 100644 --- a/packages/kernl/src/tool/sys/wakeup.ts +++ b/packages/kernl/src/tool/sys/sleep.ts @@ -1,8 +1,8 @@ /** - * Wakeup system toolkit. - * /packages/kernl/src/tool/sys/wakeup.ts + * Sleep system toolkit. + * /packages/kernl/src/tool/sys/sleep.ts * - * Provides a tool for agents to schedule a wakeup (sleep/wait) for the current thread. + * Provides a tool for agents to sleep (pause) and schedule a wakeup for the current thread. * * The thread runtime is responsible for: * - Passing the current thread_id into this tool's parameters. @@ -135,8 +135,8 @@ const wait = tool({ // --- Toolkit --- -export const wakeup = new Toolkit({ - id: "sys.wakeup", - description: "Tools for scheduling and managing agent wakeups.", +export const sleep = new Toolkit({ + id: "sys.sleep", + description: "Tools for pausing agents and scheduling wakeups.", tools: [wait], }); From 4b35008f4340e121685762d8fddea28e35fef9be Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Mon, 8 Dec 2025 16:27:32 -0700 Subject: [PATCH 11/25] Refactor sleep tool to use context threadId Updated the sleep tool to obtain threadId from the context instead of requiring it in parameters. Adjusted documentation and assertions accordingly, and ensured threadId is set in context during thread execution. --- packages/kernl/src/context.ts | 6 ++++++ packages/kernl/src/thread/thread.ts | 3 +++ packages/kernl/src/tool/sys/sleep.ts | 25 +++++++++---------------- 3 files changed, 18 insertions(+), 16 deletions(-) diff --git a/packages/kernl/src/context.ts b/packages/kernl/src/context.ts index f4c3be4a..0da468f7 100644 --- a/packages/kernl/src/context.ts +++ b/packages/kernl/src/context.ts @@ -30,6 +30,12 @@ export class Context { */ agent?: Agent; + /** + * The thread ID for the current execution. + * Set by the thread during tool execution. + */ + threadId?: string; + // ---------------------- // TEMPORARY: Tool approval tracking until actions system is refined // ---------------------- diff --git a/packages/kernl/src/thread/thread.ts b/packages/kernl/src/thread/thread.ts index 7d2302cc..9a479751 100644 --- a/packages/kernl/src/thread/thread.ts +++ b/packages/kernl/src/thread/thread.ts @@ -144,6 +144,7 @@ export class Thread< } } + // MARK: Execute /** * Blocking execution - runs until terminal state or interruption */ @@ -195,6 +196,7 @@ export class Thread< } } + // MARK: _execute main loop /** * Main execution loop - always yields events, callers can propagate or discard. * @@ -449,6 +451,7 @@ export class Thread< // is refined const ctx = new Context(this.namespace, this.context.context); ctx.agent = this.agent; + ctx.threadId = this.tid; ctx.approve(call.callId); // mark this call as approved const res = await tool.invoke(ctx, call.arguments, call.callId); diff --git a/packages/kernl/src/tool/sys/sleep.ts b/packages/kernl/src/tool/sys/sleep.ts index fdf096d5..14dcf44c 100644 --- a/packages/kernl/src/tool/sys/sleep.ts +++ b/packages/kernl/src/tool/sys/sleep.ts @@ -4,9 +4,7 @@ * * Provides a tool for agents to sleep (pause) and schedule a wakeup for the current thread. * - * The thread runtime is responsible for: - * - Passing the current thread_id into this tool's parameters. - * - Using the async/INTERRUPTIBLE tool state to checkpoint/save and stop the thread. + * The thread runtime provides threadId via ctx.threadId automatically. */ import assert from "assert"; @@ -20,18 +18,12 @@ import { Toolkit } from "../toolkit"; /** * Parameters for the wait tool. * - * The runtime must pass: - * - thread_id: the id of the thread to resume later - * And either: + * Either: * - delay_s: number of seconds from now, OR * - run_at_s: absolute epoch seconds when the thread should resume */ const WaitParamsSchema = z .object({ - thread_id: z - .string() - .min(1) - .describe("The id of the current thread to resume later."), delay_s: z .number() .int() @@ -63,14 +55,14 @@ const WaitParamsSchema = z /** * wait_until * - * Schedules a wakeup for the given thread_id and then causes the tool call + * Schedules a wakeup for the current thread and then causes the tool call * to be INTERRUPTIBLE (via requiresApproval). The actual checkpoint/save/stop * behaviour is handled by the existing async tool + thread runtime. */ const wait = tool({ id: "wait_until", description: - "Pause this agent until a future time by scheduling a wakeup for the given thread.", + "Pause this agent until a future time. The thread will be resumed automatically.", mode: "async" as const, parameters: WaitParamsSchema, @@ -79,9 +71,10 @@ const wait = tool({ * INTERRUPTIBLE and execute() is not run until some future approval/resume. */ requiresApproval: async (ctx, params) => { - assert(ctx.agent, "ctx.agent is required for wakeup tools"); + assert(ctx.agent, "ctx.agent is required for sleep tools"); + assert(ctx.threadId, "ctx.threadId is required for sleep tools"); - const { thread_id, delay_s, run_at_s, reason } = params as z.infer< + const { delay_s, run_at_s, reason } = params as z.infer< typeof WaitParamsSchema >; @@ -104,11 +97,11 @@ const wait = tool({ // Create the scheduled wakeup record. // Field names and units must match NewScheduledWakeup domain type: // - id: unique wakeup ID (generated here) - // - threadId: camelCase + // - threadId: from context (set by thread runtime) // - runAt: epoch milliseconds await wakeupStore.create({ id: `wkp_${randomID()}`, - threadId: thread_id, + threadId: ctx.threadId, runAt: targetRunAt_ms, reason: reason ?? null, }); From c9ec86f2280b856cdbd80e9953c48052fd141f27 Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Wed, 10 Dec 2025 07:57:14 -0700 Subject: [PATCH 12/25] Add sleeper agent demonstrating sleep/wakeup tool Introduces a new agent called 'Sleeper' that uses the wait_until tool to pause and wait as instructed. The agent is configured with OpenAI's GPT-5.1 model and has memory enabled. --- .../playground/server/src/agents/sleeper.ts | 15 +++++++++++++++ 1 file changed, 15 insertions(+) create mode 100644 microprojects/playground/server/src/agents/sleeper.ts diff --git a/microprojects/playground/server/src/agents/sleeper.ts b/microprojects/playground/server/src/agents/sleeper.ts new file mode 100644 index 00000000..41d7099d --- /dev/null +++ b/microprojects/playground/server/src/agents/sleeper.ts @@ -0,0 +1,15 @@ +// import "@kernl-sdk/ai/openai"; +import { openai } from "@kernl-sdk/ai/openai"; +import { Agent } from "kernl"; +// import { anthropic } from "@kernl-sdk/ai/anthropic"; + +export const sleeper = new Agent({ + id: "sleeper", + name: "Sleeper", + description: "An agent that demonstrates the sleep/wakeup system tool", + instructions: `You are a helpful assistant that can pause and wait. +When asked to wait or sleep, use the wait_until tool with an appropriate delay. +Always explain what you're doing before sleeping.`, + model: openai("gpt-5.1"), + memory: { enabled: true }, +}); From 5f51888d10b30e1f616cbc591191152c0ee78f1f Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Wed, 10 Dec 2025 07:57:39 -0700 Subject: [PATCH 13/25] Register sleeper agent in Kernl app Added the sleeper agent to the Kernl application by importing it and registering it alongside other agents. Also commented out the unused turbopuffer import. --- microprojects/playground/server/src/app.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/microprojects/playground/server/src/app.ts b/microprojects/playground/server/src/app.ts index ed17886b..a5c1d81e 100644 --- a/microprojects/playground/server/src/app.ts +++ b/microprojects/playground/server/src/app.ts @@ -1,8 +1,9 @@ import { Kernl } from "kernl"; import { pgvector, postgres } from "@kernl-sdk/pg"; -import { turbopuffer } from "@kernl-sdk/turbopuffer"; +// import { turbopuffer } from "@kernl-sdk/turbopuffer"; import { echo } from "./agents/echo"; +import { sleeper } from "./agents/sleeper"; import { titler } from "./agents/titler"; import { watson } from "./agents/watson"; @@ -16,6 +17,7 @@ export function build(): Kernl { // --- agents --- kernl.register(echo); + kernl.register(sleeper); kernl.register(titler); kernl.register(watson); From e258581f1671d61611f71573f542250f6b0f2f5f Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Wed, 10 Dec 2025 11:25:11 -0700 Subject: [PATCH 14/25] ChatGPT WiP Sleep saying Running. --- packages/kernl/src/agent.ts | 15 ++++ .../src/agent/__tests__/systools.test.ts | 86 +++++++++++++++++-- packages/kernl/src/thread/thread.ts | 23 +++-- packages/kernl/src/tool/sys/sleep.ts | 19 ++-- 4 files changed, 117 insertions(+), 26 deletions(-) diff --git a/packages/kernl/src/agent.ts b/packages/kernl/src/agent.ts index bcbba5fc..158a7f1f 100644 --- a/packages/kernl/src/agent.ts +++ b/packages/kernl/src/agent.ts @@ -276,6 +276,21 @@ export class Agent< return undefined; } + /** + * @internal + * + * Check if a tool ID belongs to a system toolkit. + * + * @param id The tool ID to check + * @returns true if the tool is a system tool + */ + isSysTool(id: string): boolean { + for (const toolkit of this.systools) { + if (toolkit.get(id)) return true; + } + return false; + } + /** * @internal * diff --git a/packages/kernl/src/agent/__tests__/systools.test.ts b/packages/kernl/src/agent/__tests__/systools.test.ts index ca2f776d..2ee458f5 100644 --- a/packages/kernl/src/agent/__tests__/systools.test.ts +++ b/packages/kernl/src/agent/__tests__/systools.test.ts @@ -26,11 +26,13 @@ describe("Agent systools", () => { const kernl = new Kernl(); kernl.register(agent); - expect(agent.systools.length).toBe(1); + // memory + sleep toolkits + expect(agent.systools.length).toBe(2); expect(agent.systools[0].id).toBe("sys.memory"); + expect(agent.systools[1].id).toBe("sys.sleep"); }); - it("has no systools when memory not configured", () => { + it("has only sleep toolkit when memory not configured", () => { const agent = new Agent({ id: "test-agent", name: "Test", @@ -41,10 +43,12 @@ describe("Agent systools", () => { const kernl = new Kernl(); kernl.register(agent); - expect(agent.systools.length).toBe(0); + // sleep is always registered + expect(agent.systools.length).toBe(1); + expect(agent.systools[0].id).toBe("sys.sleep"); }); - it("has no systools when memory.enabled is false", () => { + it("has only sleep toolkit when memory.enabled is false", () => { const agent = new Agent({ id: "test-agent", name: "Test", @@ -56,7 +60,9 @@ describe("Agent systools", () => { const kernl = new Kernl(); kernl.register(agent); - expect(agent.systools.length).toBe(0); + // sleep is always registered + expect(agent.systools.length).toBe(1); + expect(agent.systools[0].id).toBe("sys.sleep"); }); it("can retrieve memory tools via agent.tool()", () => { @@ -114,12 +120,13 @@ describe("Agent systools", () => { const ctx = new Context("test"); const tools = await agent.tools(ctx); - // Memory tools should be first (from systools) - // Order: list, create, update, search + // Memory tools should be first (from systools), then sleep tools + // Order: memory tools (list, create, update, search), then sleep tools (wait_until) expect(tools[0].id).toBe("list_memories"); expect(tools[1].id).toBe("create_memory"); expect(tools[2].id).toBe("update_memory"); expect(tools[3].id).toBe("search_memories"); + expect(tools[4].id).toBe("wait_until"); }); }); @@ -147,4 +154,69 @@ describe("Agent systools", () => { expect(agent.memory).toEqual({ enabled: true }); }); }); + + describe("sleep toolkit", () => { + it("sleep toolkit is always registered", () => { + const agent = new Agent({ + id: "test-agent", + name: "Test", + instructions: "Test", + model, + }); + + const kernl = new Kernl(); + kernl.register(agent); + + const sleepToolkit = agent.systools.find((t) => t.id === "sys.sleep"); + expect(sleepToolkit).toBeDefined(); + }); + + it("can retrieve wait_until tool via agent.tool()", () => { + const agent = new Agent({ + id: "test-agent", + name: "Test", + instructions: "Test", + model, + }); + + const kernl = new Kernl(); + kernl.register(agent); + + expect(agent.tool("wait_until")).toBeDefined(); + }); + + it("includes wait_until in agent.tools() output", async () => { + const agent = new Agent({ + id: "test-agent", + name: "Test", + instructions: "Test", + model, + }); + + const kernl = new Kernl(); + kernl.register(agent); + + const ctx = new Context("test"); + const tools = await agent.tools(ctx); + const ids = tools.map((t) => t.id); + + expect(ids).toContain("wait_until"); + }); + + it("wait_until tool has correct description", () => { + const agent = new Agent({ + id: "test-agent", + name: "Test", + instructions: "Test", + model, + }); + + const kernl = new Kernl(); + kernl.register(agent); + + const tool = agent.tool("wait_until"); + expect(tool).toBeDefined(); + expect(tool!.id).toBe("wait_until"); + }); + }); }); diff --git a/packages/kernl/src/thread/thread.ts b/packages/kernl/src/thread/thread.ts index 9a479751..2f162f8b 100644 --- a/packages/kernl/src/thread/thread.ts +++ b/packages/kernl/src/thread/thread.ts @@ -22,6 +22,7 @@ import { LanguageModel, LanguageModelItem, LanguageModelRequest, + INTERRUPTIBLE } from "@kernl-sdk/protocol"; import { randomID, filter } from "@kernl-sdk/shared/lib"; @@ -256,13 +257,9 @@ export class Thread< await this.checkpoint(); /* c3: tick complete */ if (pendingApprovals.length > 0) { - // publish a batch approval request containing all of them - // - // const reqid = randomID(); - // this.kernl.publish(channel, approvalRequest); - // - // const filter = { reqid } - // await wait_event(Action.ApprovalResponse, filter); + // Thread halts when actions require approval (e.g., sleep). + // External scheduler will resume the thread later. + return; } } } @@ -409,9 +406,11 @@ export class Thread< // (TODO): clean this - approval tracking should be handled differently for (const e of toolEvents) { + // actions.push(e); if ( e.kind === "tool-result" && - (e.state as any) === "requires_approval" // (TODO): fix this + e.state === INTERRUPTIBLE && + this.agent.isSysTool(e.toolId) ) { // find the original tool call for this pending approval const call = intentions.toolCalls.find((c) => c.callId === e.callId); @@ -452,7 +451,13 @@ export class Thread< const ctx = new Context(this.namespace, this.context.context); ctx.agent = this.agent; ctx.threadId = this.tid; - ctx.approve(call.callId); // mark this call as approved + + // Don't pre-approve system tools - they use requiresApproval to signal + // INTERRUPTIBLE state (e.g., sleep tool halts the execution loop) + if (!this.agent.isSysTool(call.toolId)) { + ctx.approve(call.callId); + } + const res = await tool.invoke(ctx, call.arguments, call.callId); return { diff --git a/packages/kernl/src/tool/sys/sleep.ts b/packages/kernl/src/tool/sys/sleep.ts index 14dcf44c..c121a1ae 100644 --- a/packages/kernl/src/tool/sys/sleep.ts +++ b/packages/kernl/src/tool/sys/sleep.ts @@ -40,9 +40,8 @@ const WaitParamsSchema = z .describe("Exact epoch seconds when this thread should be resumed."), reason: z .string() - .max(512) .optional() - .describe("Optional human-readable reason for the wakeup."), + .describe("Optional human-readable reason for the sleep."), }) .refine( (v) => (v.delay_s ?? null) !== null || (v.run_at_s ?? null) !== null, @@ -57,7 +56,7 @@ const WaitParamsSchema = z * * Schedules a wakeup for the current thread and then causes the tool call * to be INTERRUPTIBLE (via requiresApproval). The actual checkpoint/save/stop - * behaviour is handled by the existing async tool + thread runtime. + * behaviour is handled by the thread runtime. */ const wait = tool({ id: "wait_until", @@ -95,8 +94,8 @@ const wait = tool({ const targetRunAt_ms = targetRunAt_s * 1000; // Create the scheduled wakeup record. - // Field names and units must match NewScheduledWakeup domain type: - // - id: unique wakeup ID (generated here) + // Field names and units must match your NewScheduledWakeup domain type: + // - id: unique wakeup ID // - threadId: from context (set by thread runtime) // - runAt: epoch milliseconds await wakeupStore.create({ @@ -106,16 +105,16 @@ const wait = tool({ reason: reason ?? null, }); - // Returning true tells the tool engine this call requires approval, - // so it will mark the tool call as INTERRUPTIBLE and *not* run execute() - // until a scheduler/poller resumes/approves it. + // Returning true: this call requires approval → tool engine will: + // - set ToolResult.state = INTERRUPTIBLE + // - NOT call execute() in this tick return true; }, /** * execute() is only expected to run if/when the tool call is explicitly - * approved/resumed by your scheduler. For now it just returns a simple - * acknowledgement. + * approved/resumed by some higher-level scheduler. For sleep we can just + * return a simple acknowledgement; in many flows this will never be hit. */ execute: async () => { return { From 55faa31de72767285835205031836a13583f3882 Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Wed, 10 Dec 2025 11:50:32 -0700 Subject: [PATCH 15/25] Refactor thread sleep handling and approval flow Improves thread state management to distinguish between INTERRUPTIBLE and STOPPED states, ensuring correct behavior when threads are paused (e.g., sleeping). Updates the sleep system tool to use execute() for scheduling wakeups and reporting status, rather than requiresApproval, and enhances the thread event processing logic to better handle system tool interrupts and sleep events. --- packages/kernl/src/thread/thread.ts | 60 +++++++++++++++++++++------- packages/kernl/src/tool/sys/sleep.ts | 52 ++++++------------------ 2 files changed, 58 insertions(+), 54 deletions(-) diff --git a/packages/kernl/src/thread/thread.ts b/packages/kernl/src/thread/thread.ts index 2f162f8b..041e0607 100644 --- a/packages/kernl/src/thread/thread.ts +++ b/packages/kernl/src/thread/thread.ts @@ -191,9 +191,13 @@ export class Thread< } catch (err) { throw err; } finally { - this.state = STOPPED; + // Preserve INTERRUPTIBLE state (set by sleep/approval flows). + // Only transition to STOPPED if we're still RUNNING. + if (this.state === RUNNING) { + this.state = STOPPED; + } this.abort = undefined; - await this.checkpoint(); /* c4: final checkpoint - persist STOPPED state */ + await this.checkpoint(); /* c4: final checkpoint - persist final state */ } } @@ -259,6 +263,8 @@ export class Thread< if (pendingApprovals.length > 0) { // Thread halts when actions require approval (e.g., sleep). // External scheduler will resume the thread later. + // Set state to INTERRUPTIBLE so callers know the thread is sleeping, not stopped. + this.state = INTERRUPTIBLE; return; } } @@ -404,21 +410,47 @@ export class Thread< const actions: ThreadEventInner[] = []; const pendingApprovals: ToolCall[] = []; - // (TODO): clean this - approval tracking should be handled differently for (const e of toolEvents) { - // actions.push(e); - if ( - e.kind === "tool-result" && - e.state === INTERRUPTIBLE && - this.agent.isSysTool(e.toolId) - ) { - // find the original tool call for this pending approval - const call = intentions.toolCalls.find((c) => c.callId === e.callId); - call && pendingApprovals.push(call); - } else { - actions.push(e); + actions.push(e); // always record the tool call in history + + // Only system tools are allowed to affect control flow. + if (e.kind === "tool-result" && this.agent.isSysTool(e.toolId)) { + // 1) Future-proof: generic approval-based interrupts (other sys tools) + // If you later add real approval flows that still use INTERRUPTIBLE, + // this branch continues to work. + if (e.state === INTERRUPTIBLE) { + const call = intentions.toolCalls.find((c) => c.callId === e.callId); + if (call) pendingApprovals.push(call); + continue; + } + + // 2) Sleep tool: state is COMPLETED, but semantically it means + // "alarm set, thread should pause until wakeup." + if (e.toolId === "wait_until") { + const call = intentions.toolCalls.find((c) => c.callId === e.callId); + if (call) pendingApprovals.push(call); + continue; + } } } + // const actions: ThreadEventInner[] = []; + // const pendingApprovals: ToolCall[] = []; + + // // (TODO): clean this - approval tracking should be handled differently + // for (const e of toolEvents) { + // // actions.push(e); + // if ( + // e.kind === "tool-result" && + // e.state === INTERRUPTIBLE && + // this.agent.isSysTool(e.toolId) + // ) { + // // find the original tool call for this pending approval + // const call = intentions.toolCalls.find((c) => c.callId === e.callId); + // call && pendingApprovals.push(call); + // } else { + // actions.push(e); + // } + // } return { actions: actions, diff --git a/packages/kernl/src/tool/sys/sleep.ts b/packages/kernl/src/tool/sys/sleep.ts index c121a1ae..6fb7364b 100644 --- a/packages/kernl/src/tool/sys/sleep.ts +++ b/packages/kernl/src/tool/sys/sleep.ts @@ -1,10 +1,6 @@ /** * Sleep system toolkit. * /packages/kernl/src/tool/sys/sleep.ts - * - * Provides a tool for agents to sleep (pause) and schedule a wakeup for the current thread. - * - * The thread runtime provides threadId via ctx.threadId automatically. */ import assert from "assert"; @@ -15,13 +11,6 @@ import { randomID } from "@kernl-sdk/shared/lib"; import { tool } from "../tool"; import { Toolkit } from "../toolkit"; -/** - * Parameters for the wait tool. - * - * Either: - * - delay_s: number of seconds from now, OR - * - run_at_s: absolute epoch seconds when the thread should resume - */ const WaitParamsSchema = z .object({ delay_s: z @@ -51,13 +40,6 @@ const WaitParamsSchema = z }, ); -/** - * wait_until - * - * Schedules a wakeup for the current thread and then causes the tool call - * to be INTERRUPTIBLE (via requiresApproval). The actual checkpoint/save/stop - * behaviour is handled by the thread runtime. - */ const wait = tool({ id: "wait_until", description: @@ -65,11 +47,13 @@ const wait = tool({ mode: "async" as const, parameters: WaitParamsSchema, - /** - * We do the scheduling here and return true so the tool call becomes - * INTERRUPTIBLE and execute() is not run until some future approval/resume. - */ - requiresApproval: async (ctx, params) => { + // NOTE: we rely on execute(), not requiresApproval, so the tool + // ends up with state=COMPLETED (green) in the UI. + // + // This function: + // 1) Schedules the wakeup in storage. + // 2) Returns metadata so the UI can render a 'Sleeping' success state. + execute: async (ctx, params) => { assert(ctx.agent, "ctx.agent is required for sleep tools"); assert(ctx.threadId, "ctx.threadId is required for sleep tools"); @@ -93,11 +77,6 @@ const wait = tool({ run_at_s ?? (delay_s !== undefined ? now_s + delay_s : now_s); const targetRunAt_ms = targetRunAt_s * 1000; - // Create the scheduled wakeup record. - // Field names and units must match your NewScheduledWakeup domain type: - // - id: unique wakeup ID - // - threadId: from context (set by thread runtime) - // - runAt: epoch milliseconds await wakeupStore.create({ id: `wkp_${randomID()}`, threadId: ctx.threadId, @@ -105,20 +84,13 @@ const wait = tool({ reason: reason ?? null, }); - // Returning true: this call requires approval → tool engine will: - // - set ToolResult.state = INTERRUPTIBLE - // - NOT call execute() in this tick - return true; - }, - - /** - * execute() is only expected to run if/when the tool call is explicitly - * approved/resumed by some higher-level scheduler. For sleep we can just - * return a simple acknowledgement; in many flows this will never be hit. - */ - execute: async () => { + // This object becomes the `result` in the tool-result event. + // The UI can key off `status: "sleeping"` to show a green "Sleeping" badge. return { + status: "sleeping", scheduled: true, + run_at_s: targetRunAt_s, + run_at_ms: targetRunAt_ms, message: "Wakeup scheduled; the thread will resume when the wakeup is processed.", }; From fcf4999b73ec9d926d6034ba26d2b02d860b6f3b Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Wed, 10 Dec 2025 12:15:08 -0700 Subject: [PATCH 16/25] Add WakeupScheduler for automated wakeup polling Introduces the WakeupScheduler class and related types to enable automated polling and processing of due wakeups. Integrates the scheduler into Kernl, allowing configuration via KernlOptions and exposing scheduler state and controls. This facilitates automatic resumption of sleeping threads based on scheduled wakeups. --- packages/kernl/src/index.ts | 4 + packages/kernl/src/kernl/kernl.ts | 11 ++ packages/kernl/src/kernl/types.ts | 13 ++ packages/kernl/src/scheduler/index.ts | 7 + packages/kernl/src/scheduler/scheduler.ts | 183 ++++++++++++++++++++++ packages/kernl/src/scheduler/types.ts | 38 +++++ 6 files changed, 256 insertions(+) create mode 100644 packages/kernl/src/scheduler/index.ts create mode 100644 packages/kernl/src/scheduler/scheduler.ts create mode 100644 packages/kernl/src/scheduler/types.ts diff --git a/packages/kernl/src/index.ts b/packages/kernl/src/index.ts index aeab93fe..cb8a31b9 100644 --- a/packages/kernl/src/index.ts +++ b/packages/kernl/src/index.ts @@ -90,3 +90,7 @@ export type { ScheduledWakeup, ScheduledWakeupUpdate, } from "./wakeup"; + +// --- scheduler --- +export { WakeupScheduler } from "./scheduler"; +export type { WakeupSchedulerOptions, WakeupSchedulerState } from "./scheduler"; diff --git a/packages/kernl/src/kernl/kernl.ts b/packages/kernl/src/kernl/kernl.ts index 14ae3b45..b55affb2 100644 --- a/packages/kernl/src/kernl/kernl.ts +++ b/packages/kernl/src/kernl/kernl.ts @@ -18,6 +18,7 @@ import { MemoryIndexHandle, buildMemoryIndexSchema, } from "@/memory"; +import { WakeupScheduler } from "@/scheduler"; import type { ThreadExecuteResult, ThreadStreamEvent } from "@/thread/types"; import type { AgentOutputType } from "@/agent/types"; @@ -41,6 +42,7 @@ export class Kernl extends KernlHooks { readonly threads: RThreads; readonly agents: RAgents; readonly memories: Memory; + readonly scheduler: WakeupScheduler | null; constructor(options: KernlOptions = {}) { super(); @@ -76,6 +78,15 @@ export class Kernl extends KernlHooks { : undefined, encoder, }); + + // initialize scheduler + if (options.scheduler) { + const schedulerOpts = + typeof options.scheduler === "boolean" ? {} : options.scheduler; + this.scheduler = new WakeupScheduler(this, schedulerOpts); + } else { + this.scheduler = null; + } } /** diff --git a/packages/kernl/src/kernl/types.ts b/packages/kernl/src/kernl/types.ts index 07514b8a..3bf252d0 100644 --- a/packages/kernl/src/kernl/types.ts +++ b/packages/kernl/src/kernl/types.ts @@ -6,6 +6,7 @@ import { SearchIndex } from "@kernl-sdk/retrieval"; import { Agent } from "@/agent"; import { KernlStorage } from "@/storage"; +import type { WakeupSchedulerOptions } from "@/scheduler/types"; /** * Storage configuration for Kernl. @@ -84,6 +85,18 @@ export interface KernlOptions { * Memory system configuration. */ memory?: MemoryOptions; + + /** + * Scheduler configuration for wakeup polling. + * + * - If `true`, creates a scheduler with default options (30s interval, batch size 10) + * - If an object, creates a scheduler with the provided options + * - If `false` or omitted, no scheduler is created + * + * The scheduler polls for due wakeups and resumes sleeping threads. + * Call `kernl.scheduler.start()` to begin polling. + */ + scheduler?: boolean | WakeupSchedulerOptions; } /** diff --git a/packages/kernl/src/scheduler/index.ts b/packages/kernl/src/scheduler/index.ts new file mode 100644 index 00000000..90a10254 --- /dev/null +++ b/packages/kernl/src/scheduler/index.ts @@ -0,0 +1,7 @@ +/** + * Scheduler Module + * /packages/kernl/src/scheduler/index.ts + */ + +export { WakeupScheduler } from "./scheduler"; +export type { WakeupSchedulerOptions, WakeupSchedulerState } from "./types"; diff --git a/packages/kernl/src/scheduler/scheduler.ts b/packages/kernl/src/scheduler/scheduler.ts new file mode 100644 index 00000000..a17dc3cd --- /dev/null +++ b/packages/kernl/src/scheduler/scheduler.ts @@ -0,0 +1,183 @@ +/** + * Wakeup Scheduler + * /packages/kernl/src/scheduler/scheduler.ts + * + * Polls the wakeup store for due wakeups and resumes threads. + * Can be attached to a Kernl instance or run standalone. + */ + +import type { Kernl } from "@/kernl"; +import type { ScheduledWakeup } from "@/wakeup"; +import { getLogger } from "@/lib/logger"; + +import type { WakeupSchedulerOptions, WakeupSchedulerState } from "./types"; + +const logger = getLogger("kernl:scheduler"); + +const DEFAULT_INTERVAL_MS = 30_000; // 30 seconds +const DEFAULT_BATCH_SIZE = 10; + +export class WakeupScheduler { + private readonly kernl: Kernl; + private readonly intervalMs: number; + private readonly batchSize: number; + + private timer: ReturnType | null = null; + private polling = false; // guard against overlapping polls + + private _state: WakeupSchedulerState = { + running: false, + processed: 0, + failed: 0, + lastPollAt: null, + }; + + constructor(kernl: Kernl, options: WakeupSchedulerOptions = {}) { + this.kernl = kernl; + this.intervalMs = options.intervalMs ?? DEFAULT_INTERVAL_MS; + this.batchSize = options.batchSize ?? DEFAULT_BATCH_SIZE; + + if (options.autoStart) { + this.start(); + } + } + + /** + * Current scheduler state (read-only snapshot). + */ + get state(): Readonly { + return { ...this._state }; + } + + /** + * Start the polling loop. + * No-op if already running. + */ + start(): void { + if (this._state.running) { + logger.warn("scheduler already running"); + return; + } + + logger.info({ intervalMs: this.intervalMs }, "scheduler starting"); + this._state.running = true; + + // Run first poll immediately, then on interval + this.poll(); + this.timer = setInterval(() => this.poll(), this.intervalMs); + } + + /** + * Stop the polling loop. + * In-flight poll will complete but no new polls will start. + */ + stop(): void { + if (!this._state.running) { + logger.warn("scheduler not running"); + return; + } + + logger.info("scheduler stopping"); + this._state.running = false; + + if (this.timer) { + clearInterval(this.timer); + this.timer = null; + } + } + + /** + * Execute a single poll iteration. + * Claims due wakeups and resumes their threads. + * + * Can be called manually for testing or one-off processing. + */ + async poll(): Promise { + // Guard against overlapping polls (previous poll still running) + if (this.polling) { + logger.debug("skipping poll, previous poll still running"); + return; + } + + this.polling = true; + this._state.lastPollAt = Date.now(); + + try { + const wakeupStore = this.kernl.storage?.wakeups; + if (!wakeupStore) { + logger.warn("no wakeup store configured, skipping poll"); + return; + } + + const nowMs = Date.now(); + const claimed = await wakeupStore.claimDue(nowMs, this.batchSize); + + if (claimed.length === 0) { + logger.debug("no due wakeups found"); + return; + } + + logger.info({ count: claimed.length }, "claimed due wakeups"); + + // Process each wakeup + await Promise.all(claimed.map((wakeup) => this.processWakeup(wakeup))); + } catch (err) { + logger.error({ err }, "poll failed"); + } finally { + this.polling = false; + } + } + + /** + * Process a single wakeup: resume the thread and mark wakeup as woken. + */ + private async processWakeup(wakeup: ScheduledWakeup): Promise { + const { id, threadId } = wakeup; + const wakeupStore = this.kernl.storage?.wakeups; + const threadStore = this.kernl.storage?.threads; + + if (!wakeupStore || !threadStore) { + logger.error({ wakeupId: id }, "storage not configured"); + return; + } + + try { + logger.debug({ wakeupId: id, threadId }, "processing wakeup"); + + // Load thread with history + const thread = await threadStore.get(threadId, { history: true }); + + if (!thread) { + throw new Error(`Thread ${threadId} not found`); + } + + // Resume thread execution (thread.execute continues from where it left off) + await this.kernl.schedule(thread); + + // Mark wakeup as complete + await wakeupStore.update(id, { + id, + woken: true, + updatedAt: Date.now(), + }); + + this._state.processed++; + logger.info({ wakeupId: id, threadId }, "wakeup processed successfully"); + } catch (err) { + this._state.failed++; + const errorMsg = err instanceof Error ? err.message : String(err); + logger.error({ wakeupId: id, threadId, err: errorMsg }, "wakeup processing failed"); + + // Record error on the wakeup record + try { + await wakeupStore.update(id, { + id, + error: errorMsg, + updatedAt: Date.now(), + }); + } catch (updateErr) { + logger.error({ wakeupId: id, err: updateErr }, "failed to update wakeup with error"); + } + } + } +} diff --git a/packages/kernl/src/scheduler/types.ts b/packages/kernl/src/scheduler/types.ts new file mode 100644 index 00000000..68edf835 --- /dev/null +++ b/packages/kernl/src/scheduler/types.ts @@ -0,0 +1,38 @@ +/** + * Scheduler Types + * /packages/kernl/src/scheduler/types.ts + */ + +export interface WakeupSchedulerOptions { + /** + * Polling interval in milliseconds. + * @default 30000 (30 seconds) + */ + intervalMs?: number; + + /** + * Maximum number of wakeups to claim per poll. + * @default 10 + */ + batchSize?: number; + + /** + * Whether to start polling automatically when the scheduler is created. + * @default false + */ + autoStart?: boolean; +} + +export interface WakeupSchedulerState { + /** Whether the scheduler is currently polling */ + running: boolean; + + /** Number of wakeups processed since start */ + processed: number; + + /** Number of wakeups that failed to process */ + failed: number; + + /** Timestamp of last poll (epoch ms), null if never polled */ + lastPollAt: number | null; +} From 4771d8d900b87cd531f43db54401de89a13c39ec Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Wed, 10 Dec 2025 12:40:34 -0700 Subject: [PATCH 17/25] Switch kernl dependency to workspace and enable scheduler Updated kernl dependency in package.json to use workspace reference. Enabled and started the scheduler in app.ts, and performed minor formatting cleanup in sleeper.ts. --- microprojects/playground/server/package.json | 2 +- microprojects/playground/server/src/agents/sleeper.ts | 1 + microprojects/playground/server/src/app.ts | 6 ++++++ 3 files changed, 8 insertions(+), 1 deletion(-) diff --git a/microprojects/playground/server/package.json b/microprojects/playground/server/package.json index eda79426..68f2757d 100644 --- a/microprojects/playground/server/package.json +++ b/microprojects/playground/server/package.json @@ -13,7 +13,7 @@ "@kernl-sdk/pg": "workspace:*", "@kernl-sdk/server": "workspace:*", "@kernl-sdk/turbopuffer": "workspace:*", - "kernl": "*" + "kernl": "workspace:*" }, "devDependencies": { "@types/node": "^24.10.0", diff --git a/microprojects/playground/server/src/agents/sleeper.ts b/microprojects/playground/server/src/agents/sleeper.ts index 41d7099d..aa4ec562 100644 --- a/microprojects/playground/server/src/agents/sleeper.ts +++ b/microprojects/playground/server/src/agents/sleeper.ts @@ -3,6 +3,7 @@ import { openai } from "@kernl-sdk/ai/openai"; import { Agent } from "kernl"; // import { anthropic } from "@kernl-sdk/ai/anthropic"; + export const sleeper = new Agent({ id: "sleeper", name: "Sleeper", diff --git a/microprojects/playground/server/src/app.ts b/microprojects/playground/server/src/app.ts index a5c1d81e..605e88d4 100644 --- a/microprojects/playground/server/src/app.ts +++ b/microprojects/playground/server/src/app.ts @@ -1,5 +1,7 @@ import { Kernl } from "kernl"; import { pgvector, postgres } from "@kernl-sdk/pg"; +import "@kernl-sdk/ai/openai"; + // import { turbopuffer } from "@kernl-sdk/turbopuffer"; import { echo } from "./agents/echo"; @@ -13,6 +15,7 @@ export function build(): Kernl { db: postgres({ connstr: process.env.DATABASE_URL! }), vector: pgvector({ connstr: process.env.DATABASE_URL! }), }, + scheduler: true, }); // --- agents --- @@ -21,6 +24,9 @@ export function build(): Kernl { kernl.register(titler); kernl.register(watson); + // start wakeup scheduler + kernl.schedule?.start(); + return kernl; } From ae588c4780a9f20fcdf290ec767f491526826d33 Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Wed, 10 Dec 2025 13:39:29 -0700 Subject: [PATCH 18/25] Update scheduler configuration in app setup Replaces the boolean scheduler option with an object specifying autoStart. Also comments out the explicit scheduler start call, likely to rely on the new autoStart behavior. --- microprojects/playground/server/src/app.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/microprojects/playground/server/src/app.ts b/microprojects/playground/server/src/app.ts index 605e88d4..025bfee0 100644 --- a/microprojects/playground/server/src/app.ts +++ b/microprojects/playground/server/src/app.ts @@ -15,7 +15,10 @@ export function build(): Kernl { db: postgres({ connstr: process.env.DATABASE_URL! }), vector: pgvector({ connstr: process.env.DATABASE_URL! }), }, - scheduler: true, + // scheduler: true, + scheduler: { + autoStart: true + } }); // --- agents --- @@ -25,7 +28,7 @@ export function build(): Kernl { kernl.register(watson); // start wakeup scheduler - kernl.schedule?.start(); + // kernl.schedule?.start(); return kernl; } From 0bcf44083218151af901df6de92501ec74f2847a Mon Sep 17 00:00:00 2001 From: ConnorIllingworth <46832931+ConnorIllingworth@users.noreply.github.com> Date: Wed, 10 Dec 2025 13:42:48 -0700 Subject: [PATCH 19/25] Remove placeholder comment from agent index Deleted a TODO comment from packages/kernl/src/agent/index.ts, leaving the file empty. --- packages/kernl/src/agent/index.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/packages/kernl/src/agent/index.ts b/packages/kernl/src/agent/index.ts index a5de3b36..e69de29b 100644 --- a/packages/kernl/src/agent/index.ts +++ b/packages/kernl/src/agent/index.ts @@ -1 +0,0 @@ -// TODO: Ask Andrew, Not sure if this should be blank \ No newline at end of file From f23b3fd04efb3b1ee76e2a405d59122b227561f1 Mon Sep 17 00:00:00 2001 From: Yoconn <46832931+ConnorIllingworth@users.noreply.github.com> Date: Thu, 11 Dec 2025 21:00:04 -0700 Subject: [PATCH 20/25] Rename AGENTS.md to .codex/AGENTS.md Moved AGENTS.md to the .codex directory for better organization or to follow project structure conventions. --- AGENTS.md => .codex/AGENTS.md | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename AGENTS.md => .codex/AGENTS.md (100%) diff --git a/AGENTS.md b/.codex/AGENTS.md similarity index 100% rename from AGENTS.md rename to .codex/AGENTS.md From fd5057050d6c717744329786736d0111cf077e2e Mon Sep 17 00:00:00 2001 From: Yoconn <46832931+ConnorIllingworth@users.noreply.github.com> Date: Thu, 11 Dec 2025 21:47:08 -0700 Subject: [PATCH 21/25] Refactor scheduled wakeups schema and migration Renames `run_at_s` to `wakeup_at`, adds `sleep_for` to scheduled wakeups, and updates all related codecs, SQL, and store logic for consistency. Adds a migration to handle schema changes, backfill data, and update indexes. All timestamps are now stored in epoch seconds for stability and consistency. --- packages/storage/core/src/wakeup/schema.ts | 15 ++-- packages/storage/pg/src/migrations.ts | 86 ++++++++++++++++++++++ packages/storage/pg/src/wakeup/codec.ts | 60 +++++++++++---- packages/storage/pg/src/wakeup/sql.ts | 28 +++++-- packages/storage/pg/src/wakeup/store.ts | 16 ++-- 5 files changed, 173 insertions(+), 32 deletions(-) diff --git a/packages/storage/core/src/wakeup/schema.ts b/packages/storage/core/src/wakeup/schema.ts index bd53b8ed..6f5d601f 100644 --- a/packages/storage/core/src/wakeup/schema.ts +++ b/packages/storage/core/src/wakeup/schema.ts @@ -2,7 +2,8 @@ * /packages/storage/core/src/wakeup/schema.ts * * First implementation: - * - run_at_s: epoch seconds when the wakeup becomes due + * - wakeup_at: epoch seconds when the wakeup becomes due + * - sleep_for: seconds to sleep (requested duration) * - woken: consumed/completed * - claimed_at_s: set when a poller claims it to avoid double-processing */ @@ -21,8 +22,11 @@ export const TABLE_SCHEDULED_WAKEUPS = defineTable( onDelete: "CASCADE", }), + // Requested duration (seconds) + sleep_for: bigint(), + // Due time (epoch seconds) - run_at_s: bigint(), + wakeup_at: bigint(), reason: text().nullable(), @@ -37,8 +41,8 @@ export const TABLE_SCHEDULED_WAKEUPS = defineTable( error: text().nullable(), }, [ - // Polling query: woken=false AND claimed_at_s IS NULL AND run_at_s <= now - { kind: "index", columns: ["woken", "run_at_s"] }, + // Polling query: woken=false AND claimed_at_s IS NULL AND wakeup_at <= now + { kind: "index", columns: ["woken", "wakeup_at"] }, { kind: "index", columns: ["thread_id"] }, { kind: "index", columns: ["claimed_at_s"] }, ], @@ -50,7 +54,8 @@ export const ScheduledWakeupRecordSchema = z.object({ id: z.string(), thread_id: z.string(), - run_at_s: epochSeconds, + sleep_for: epochSeconds, + wakeup_at: epochSeconds, reason: z.string().nullable(), woken: z.boolean(), diff --git a/packages/storage/pg/src/migrations.ts b/packages/storage/pg/src/migrations.ts index 07e239d0..73e2badb 100644 --- a/packages/storage/pg/src/migrations.ts +++ b/packages/storage/pg/src/migrations.ts @@ -47,6 +47,92 @@ export const MIGRATIONS: Migration[] = [ await ctx.createTable(TABLE_SCHEDULED_WAKEUPS); }, }, + { + id: "004_scheduled_wakeups_v2", + async up(ctx) { + // v1 -> v2: + // - run_at_s -> wakeup_at + // - add sleep_for + // - convert created_at/updated_at from ms -> seconds (if needed) + // - backfill sleep_for for existing rows + const schema = "kernl"; + + const colsRes = await ctx.client.query<{ column_name: string }>( + ` + SELECT column_name + FROM information_schema.columns + WHERE table_schema = $1 AND table_name = 'scheduled_wakeups' + `, + [schema], + ); + + if (colsRes.rows.length === 0) return; + + const cols = new Set(colsRes.rows.map((r) => r.column_name)); + + if (cols.has("run_at_s") && !cols.has("wakeup_at")) { + await ctx.client.query( + `ALTER TABLE ${schema}.scheduled_wakeups RENAME COLUMN run_at_s TO wakeup_at`, + ); + cols.delete("run_at_s"); + cols.add("wakeup_at"); + } + + // If we somehow ended up with both columns, keep wakeup_at and drop the old name. + if (cols.has("run_at_s") && cols.has("wakeup_at")) { + await ctx.client.query( + `ALTER TABLE ${schema}.scheduled_wakeups DROP COLUMN run_at_s`, + ); + cols.delete("run_at_s"); + } + + if (!cols.has("sleep_for")) { + await ctx.client.query( + `ALTER TABLE ${schema}.scheduled_wakeups ADD COLUMN sleep_for bigint`, + ); + cols.add("sleep_for"); + } + + // Convert ms -> seconds only when the values are clearly ms. + await ctx.client.query( + ` + UPDATE ${schema}.scheduled_wakeups + SET created_at = created_at / 1000, + updated_at = updated_at / 1000 + WHERE created_at > 100000000000 OR updated_at > 100000000000; + `, + ); + + // Backfill sleep_for for existing rows. + await ctx.client.query( + ` + UPDATE ${schema}.scheduled_wakeups + SET sleep_for = GREATEST(0, wakeup_at - created_at) + WHERE sleep_for IS NULL; + `, + ); + + // Enforce non-null/default after backfill. + await ctx.client.query( + `ALTER TABLE ${schema}.scheduled_wakeups ALTER COLUMN sleep_for SET DEFAULT 0`, + ); + await ctx.client.query( + `ALTER TABLE ${schema}.scheduled_wakeups ALTER COLUMN sleep_for SET NOT NULL`, + ); + + // Ensure the polling indexes exist (safe even if duplicates already exist under other names). + await ctx.client.query( + `CREATE INDEX IF NOT EXISTS scheduled_wakeups_woken_wakeup_at_idx ON ${schema}.scheduled_wakeups (woken, wakeup_at)`, + ); + await ctx.client.query( + `CREATE INDEX IF NOT EXISTS scheduled_wakeups_thread_id_idx ON ${schema}.scheduled_wakeups (thread_id)`, + ); + await ctx.client.query( + `CREATE INDEX IF NOT EXISTS scheduled_wakeups_claimed_at_s_idx ON ${schema}.scheduled_wakeups (claimed_at_s)`, + ); + }, + }, + ]; /** diff --git a/packages/storage/pg/src/wakeup/codec.ts b/packages/storage/pg/src/wakeup/codec.ts index a6c92c50..803ee751 100644 --- a/packages/storage/pg/src/wakeup/codec.ts +++ b/packages/storage/pg/src/wakeup/codec.ts @@ -22,18 +22,28 @@ export const NewScheduledWakeupCodec: Codec< ScheduledWakeupRecord > = { encode(input) { - const nowMs = Date.now(); - const runAtS = Math.floor(input.runAt / 1000); + // Store timestamps in epoch *seconds* to keep the DB representation stable. + const nowS = Math.floor(Date.now() / 1000); + + // Back-compat: allow either `sleepFor` (seconds) or legacy `runAt` (ms epoch). + const anyInput = input as any; + const sleepForS = + typeof anyInput.sleepFor === "number" + ? Math.max(0, Math.floor(anyInput.sleepFor)) + : Math.max(0, Math.floor(anyInput.runAt / 1000) - nowS); + + const wakeupAtS = nowS + sleepForS; return { id: input.id, thread_id: input.threadId, - run_at_s: runAtS, + sleep_for: sleepForS, + wakeup_at: wakeupAtS, reason: input.reason ?? null, woken: false, claimed_at_s: null, - created_at: nowMs, - updated_at: nowMs, + created_at: nowS, + updated_at: nowS, error: null, }; }, @@ -51,19 +61,40 @@ export const ScheduledWakeupCodec: Codec< ScheduledWakeupRecord > = { encode(wakeup) { - const runAtS = Math.floor(wakeup.runAt / 1000); + const anyWakeup = wakeup as any; + + const wakeupAtMs = anyWakeup.wakeupAt ?? anyWakeup.runAt; + const wakeupAtS = Math.floor(wakeupAtMs / 1000); + + // Domain timestamps are typically ms; DB stores seconds. + const createdAtS = + typeof anyWakeup.createdAt === "number" + ? Math.floor(anyWakeup.createdAt / 1000) + : wakeupAtS; + + const updatedAtS = + typeof anyWakeup.updatedAt === "number" + ? Math.floor(anyWakeup.updatedAt / 1000) + : createdAtS; + + const sleepForS = + typeof anyWakeup.sleepFor === "number" + ? Math.max(0, Math.floor(anyWakeup.sleepFor)) + : Math.max(0, wakeupAtS - createdAtS); + const claimedAtS = - wakeup.claimedAt != null ? Math.floor(wakeup.claimedAt / 1000) : null; + anyWakeup.claimedAt != null ? Math.floor(anyWakeup.claimedAt / 1000) : null; return { id: wakeup.id, thread_id: wakeup.threadId, - run_at_s: runAtS, + sleep_for: sleepForS, + wakeup_at: wakeupAtS, reason: wakeup.reason, woken: wakeup.woken, claimed_at_s: claimedAtS, - created_at: wakeup.createdAt, - updated_at: wakeup.updatedAt, + created_at: createdAtS, + updated_at: updatedAtS, error: wakeup.error, }; }, @@ -72,13 +103,16 @@ export const ScheduledWakeupCodec: Codec< return { id: record.id, threadId: record.thread_id, - runAt: record.run_at_s * 1000, + // Back-compat: expose legacy `runAt` in ms while the DB stores seconds. + runAt: record.wakeup_at * 1000, + // New: also expose the stored duration (seconds). + sleepFor: record.sleep_for, reason: record.reason, woken: record.woken, claimedAt: record.claimed_at_s != null ? record.claimed_at_s * 1000 : null, - createdAt: record.created_at, - updatedAt: record.updated_at, + createdAt: record.created_at * 1000, + updatedAt: record.updated_at * 1000, error: record.error, }; }, diff --git a/packages/storage/pg/src/wakeup/sql.ts b/packages/storage/pg/src/wakeup/sql.ts index d1cb7ada..d8d0cbe8 100644 --- a/packages/storage/pg/src/wakeup/sql.ts +++ b/packages/storage/pg/src/wakeup/sql.ts @@ -30,9 +30,25 @@ export const PATCH: Codec = { const params: unknown[] = []; let idx = startIdx; - if (patch.runAt !== undefined) { - sets.push(`run_at_s = $${idx++}`); - params.push(Math.floor(patch.runAt / 1000)); + // Back-compat: allow either `wakeupAt` (ms), legacy `runAt` (ms), and/or `sleepFor` (seconds). + const anyPatch = patch as any; + const wakeupAtMs = anyPatch.wakeupAt ?? anyPatch.runAt; + + if (wakeupAtMs !== undefined) { + const wakeupAtS = Math.floor(wakeupAtMs / 1000); + const wakeupIdx = idx++; + sets.push(`wakeup_at = $${wakeupIdx}`); + params.push(wakeupAtS); + + // If caller did not explicitly set sleepFor, keep it consistent with wakeup_at and created_at. + if (anyPatch.sleepFor === undefined) { + sets.push(`sleep_for = GREATEST(0, $${wakeupIdx} - created_at)`); + } + } + + if (anyPatch.sleepFor !== undefined) { + sets.push(`sleep_for = $${idx++}`); + params.push(Math.max(0, Math.floor(anyPatch.sleepFor))); } if (patch.reason !== undefined) { @@ -57,10 +73,10 @@ export const PATCH: Codec = { params.push(patch.error); } - // Always bump updated_at in ms, matching memory's PATCH behavior. - const nowMs = patch.updatedAt ?? Date.now(); + // Always bump updated_at in epoch seconds. + const nowS = Math.floor(((patch as any).updatedAt ?? Date.now()) / 1000); sets.push(`updated_at = $${idx++}`); - params.push(nowMs); + params.push(nowS); return { sql: sets.join(", "), diff --git a/packages/storage/pg/src/wakeup/store.ts b/packages/storage/pg/src/wakeup/store.ts index fb746755..bb722716 100644 --- a/packages/storage/pg/src/wakeup/store.ts +++ b/packages/storage/pg/src/wakeup/store.ts @@ -57,13 +57,14 @@ export class PGWakeupStore implements WakeupStore { const result = await this.db.query( `INSERT INTO ${KERNL_SCHEMA_NAME}.scheduled_wakeups - (id, thread_id, run_at_s, reason, woken, claimed_at_s, created_at, updated_at, error) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) + (id, thread_id, sleep_for, wakeup_at, reason, woken, claimed_at_s, created_at, updated_at, error) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) RETURNING *`, [ row.id, row.thread_id, - row.run_at_s, + row.sleep_for, + row.wakeup_at, row.reason, row.woken, row.claimed_at_s, @@ -135,7 +136,6 @@ export class PGWakeupStore implements WakeupStore { ): Promise { await this.ensureInit(); - const nowMsNum = toMs(nowMs); const nowS = toSeconds(nowMs); const result = await this.db.query( @@ -145,19 +145,19 @@ export class PGWakeupStore implements WakeupStore { FROM ${KERNL_SCHEMA_NAME}.scheduled_wakeups WHERE woken = FALSE AND claimed_at_s IS NULL - AND run_at_s <= $1 - ORDER BY run_at_s ASC + AND wakeup_at <= $1 + ORDER BY wakeup_at ASC LIMIT $2 FOR UPDATE SKIP LOCKED ) UPDATE ${KERNL_SCHEMA_NAME}.scheduled_wakeups AS sw SET claimed_at_s = $1, - updated_at = $3 + updated_at = $1 FROM due WHERE sw.id = due.id RETURNING sw.*; `, - [nowS, limit, nowMsNum], + [nowS, limit], ); return result.rows.map((row) => From 8bb6aab6fc8d0f6c035e5a0af415c945ab18fef0 Mon Sep 17 00:00:00 2001 From: Yoconn <46832931+ConnorIllingworth@users.noreply.github.com> Date: Thu, 11 Dec 2025 22:05:06 -0700 Subject: [PATCH 22/25] Refactor wakeup scheduling to use sleepFor duration Updated wakeup scheduling logic to prefer a new `sleepFor` (seconds) field over the legacy `runAt` (epoch ms) for specifying wakeup times. Adjusted in-memory store, sleep tool, and types to support both fields for backward compatibility, with `sleepFor` as the primary method going forward. --- packages/kernl/src/storage/in-memory.ts | 14 +++++++++++++- packages/kernl/src/tool/sys/sleep.ts | 13 ++++++++++--- packages/kernl/src/wakeup/types.ts | 17 +++++++++++++++-- 3 files changed, 38 insertions(+), 6 deletions(-) diff --git a/packages/kernl/src/storage/in-memory.ts b/packages/kernl/src/storage/in-memory.ts index e51b6b6b..a5970f33 100644 --- a/packages/kernl/src/storage/in-memory.ts +++ b/packages/kernl/src/storage/in-memory.ts @@ -562,11 +562,23 @@ export class InMemoryWakeupStore implements WakeupStore { async create(input: NewScheduledWakeup): Promise { const now = Date.now(); + const nowS = Math.floor(now / 1000); + + // Support both new (sleepFor) and legacy (runAt) fields. + const sleepForS = + typeof input.sleepFor === "number" + ? Math.max(0, Math.floor(input.sleepFor)) + : typeof input.runAt === "number" + ? Math.max(0, Math.floor(input.runAt / 1000) - nowS) + : 0; + + const runAt = nowS * 1000 + sleepForS * 1000; const record: ScheduledWakeup = { id: input.id, threadId: input.threadId, - runAt: input.runAt, + runAt, + sleepFor: sleepForS, reason: input.reason ?? null, woken: false, claimedAt: null, diff --git a/packages/kernl/src/tool/sys/sleep.ts b/packages/kernl/src/tool/sys/sleep.ts index 6fb7364b..dc52381b 100644 --- a/packages/kernl/src/tool/sys/sleep.ts +++ b/packages/kernl/src/tool/sys/sleep.ts @@ -73,14 +73,21 @@ const wait = tool({ } const now_s = Math.floor(Date.now() / 1000); - const targetRunAt_s = - run_at_s ?? (delay_s !== undefined ? now_s + delay_s : now_s); + + // Calculate sleep duration in seconds. + // If delay_s is provided, use it directly. + // If run_at_s is provided, calculate the duration from now. + const sleepForS = + delay_s !== undefined ? delay_s : Math.max(0, (run_at_s ?? now_s) - now_s); + + // For UI display purposes + const targetRunAt_s = now_s + sleepForS; const targetRunAt_ms = targetRunAt_s * 1000; await wakeupStore.create({ id: `wkp_${randomID()}`, threadId: ctx.threadId, - runAt: targetRunAt_ms, + sleepFor: sleepForS, reason: reason ?? null, }); diff --git a/packages/kernl/src/wakeup/types.ts b/packages/kernl/src/wakeup/types.ts index 32785cd0..5be32875 100644 --- a/packages/kernl/src/wakeup/types.ts +++ b/packages/kernl/src/wakeup/types.ts @@ -15,12 +15,20 @@ export interface NewScheduledWakeup { threadId: string; /** - * When this wakeup becomes due (epoch milliseconds). + * Duration to sleep (seconds). Preferred over `runAt`. * + * The system will calculate the absolute wakeup time as `created_at + sleepFor`. + */ + sleepFor?: number; + + /** + * (Legacy) When this wakeup becomes due (epoch milliseconds). + * + * For backward compatibility. Prefer using `sleepFor` instead. * Storage providers can store this as seconds (`run_at_s`) or ms, * codecs handle conversion. */ - runAt: number; + runAt?: number; /** * Optional free-form reason for observability. @@ -40,6 +48,11 @@ export interface ScheduledWakeup { */ runAt: number; + /** + * Duration to sleep (seconds). + */ + sleepFor: number; + reason: string | null; /** From c04bfb6bed4967ab7d027f11485187ad43de2a2e Mon Sep 17 00:00:00 2001 From: Yoconn <46832931+ConnorIllingworth@users.noreply.github.com> Date: Thu, 11 Dec 2025 22:12:45 -0700 Subject: [PATCH 23/25] Simplify wait tool to require delay_s parameter Removed support for run_at_s in the WaitParamsSchema and related logic, making delay_s a required parameter for specifying sleep duration. This streamlines the wait tool's interface and reduces complexity. --- packages/kernl/src/tool/sys/sleep.ts | 55 ++++++++-------------------- 1 file changed, 16 insertions(+), 39 deletions(-) diff --git a/packages/kernl/src/tool/sys/sleep.ts b/packages/kernl/src/tool/sys/sleep.ts index dc52381b..4381bad3 100644 --- a/packages/kernl/src/tool/sys/sleep.ts +++ b/packages/kernl/src/tool/sys/sleep.ts @@ -11,34 +11,19 @@ import { randomID } from "@kernl-sdk/shared/lib"; import { tool } from "../tool"; import { Toolkit } from "../toolkit"; -const WaitParamsSchema = z - .object({ - delay_s: z - .number() - .int() - .nonnegative() - .optional() - .describe( - "How many seconds from now to wait before resuming this thread.", - ), - run_at_s: z - .number() - .int() - .nonnegative() - .optional() - .describe("Exact epoch seconds when this thread should be resumed."), - reason: z - .string() - .optional() - .describe("Optional human-readable reason for the sleep."), - }) - .refine( - (v) => (v.delay_s ?? null) !== null || (v.run_at_s ?? null) !== null, - { - message: "Either delay_s or run_at_s must be provided", - path: ["delay_s"], - }, - ); +const WaitParamsSchema = z.object({ + delay_s: z + .number() + .int() + .nonnegative() + .describe( + "How many seconds from now to wait before resuming this thread.", + ), + reason: z + .string() + .optional() + .describe("Optional human-readable reason for the sleep."), +}); const wait = tool({ id: "wait_until", @@ -57,9 +42,7 @@ const wait = tool({ assert(ctx.agent, "ctx.agent is required for sleep tools"); assert(ctx.threadId, "ctx.threadId is required for sleep tools"); - const { delay_s, run_at_s, reason } = params as z.infer< - typeof WaitParamsSchema - >; + const { delay_s, reason } = params as z.infer; const agent: any = ctx.agent; @@ -74,20 +57,14 @@ const wait = tool({ const now_s = Math.floor(Date.now() / 1000); - // Calculate sleep duration in seconds. - // If delay_s is provided, use it directly. - // If run_at_s is provided, calculate the duration from now. - const sleepForS = - delay_s !== undefined ? delay_s : Math.max(0, (run_at_s ?? now_s) - now_s); - // For UI display purposes - const targetRunAt_s = now_s + sleepForS; + const targetRunAt_s = now_s + delay_s; const targetRunAt_ms = targetRunAt_s * 1000; await wakeupStore.create({ id: `wkp_${randomID()}`, threadId: ctx.threadId, - sleepFor: sleepForS, + sleepFor: delay_s, reason: reason ?? null, }); From 80a8ffb0502c45ef11ee05f76151e7cee29b2ebe Mon Sep 17 00:00:00 2001 From: Yoconn <46832931+ConnorIllingworth@users.noreply.github.com> Date: Thu, 11 Dec 2025 22:22:53 -0700 Subject: [PATCH 24/25] Add migration to enable vector extension Introduces a new migration (000_enable_vector) that ensures the 'vector' extension is enabled in the database before other migrations are applied. --- packages/storage/pg/src/migrations.ts | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/packages/storage/pg/src/migrations.ts b/packages/storage/pg/src/migrations.ts index 73e2badb..4e064a26 100644 --- a/packages/storage/pg/src/migrations.ts +++ b/packages/storage/pg/src/migrations.ts @@ -28,6 +28,12 @@ export interface Migration { * List of all migrations in order. */ export const MIGRATIONS: Migration[] = [ + { + id: "000_enable_vector", + async up(ctx) { + await ctx.client.query("CREATE EXTENSION IF NOT EXISTS vector"); + }, + }, { id: "001_threads", async up(ctx) { From 5f0fd8c15ebcbe780c547198b16843fcea12d87b Mon Sep 17 00:00:00 2001 From: Yoconn <46832931+ConnorIllingworth@users.noreply.github.com> Date: Thu, 11 Dec 2025 23:06:05 -0700 Subject: [PATCH 25/25] Add user message on thread wakeup in scheduler When a thread wakes up, a user message is appended to the thread indicating the sleep period is complete and the task should continue. This provides clearer context for thread resumption. --- packages/kernl/src/scheduler/scheduler.ts | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/packages/kernl/src/scheduler/scheduler.ts b/packages/kernl/src/scheduler/scheduler.ts index a17dc3cd..aa647ea7 100644 --- a/packages/kernl/src/scheduler/scheduler.ts +++ b/packages/kernl/src/scheduler/scheduler.ts @@ -9,6 +9,7 @@ import type { Kernl } from "@/kernl"; import type { ScheduledWakeup } from "@/wakeup"; import { getLogger } from "@/lib/logger"; +import { message } from "@kernl-sdk/protocol"; import type { WakeupSchedulerOptions, WakeupSchedulerState } from "./types"; @@ -151,6 +152,18 @@ export class WakeupScheduler { throw new Error(`Thread ${threadId} not found`); } + // Append a user message to signal wakeup and continuation + const wakeupMessage = wakeup.reason + ? `[Sleep period complete - reason: ${wakeup.reason}] You have woken up. Continue with your task.` + : "[Sleep period complete] You have woken up. Continue with your task."; + + thread.append( + message({ + role: "user", + text: wakeupMessage, + }), + ); + // Resume thread execution (thread.execute continues from where it left off) await this.kernl.schedule(thread);