diff --git a/docs/contributing/architecture/request-lifecycle.md b/docs/contributing/architecture/request-lifecycle.md index 5970d8fa9b..a96ab35f53 100644 --- a/docs/contributing/architecture/request-lifecycle.md +++ b/docs/contributing/architecture/request-lifecycle.md @@ -54,6 +54,20 @@ Requests are handled in this order: 9. App server routes: - Everything else is handled by `packages/worker/src/app/handler.ts` +## Package service runtime + +Saved packages may also declare long-lived package services under +`package.json#kody.services`. + +- Package services are **not** routed through a public HTTP path the way package + apps are. +- Instead, the Worker hosts them via the `PackageServiceInstance` Durable Object + binding and controls them through package runtime bridges and MCP + capabilities. +- Package services share package identity with apps/jobs and can publish into + package app realtime sessions, but they own their own durable storage bucket + and lifecycle. + ## App server flow `packages/worker/src/app/handler.ts` validates environment variables and diff --git a/docs/guides/README.md b/docs/guides/README.md index 9d1f64331b..52ada56626 100644 --- a/docs/guides/README.md +++ b/docs/guides/README.md @@ -10,6 +10,7 @@ available `guide` ids). | [integration-bootstrap.md](./integration-bootstrap.md) | **Start here** for third-party integrations that must work before saving a dependent package or package app | | [secret-backed-integration.md](./secret-backed-integration.md) | Default recipe for non-OAuth integrations that use one or more saved secrets | | [integration-backed-app-happy-path.md](./integration-backed-app-happy-path.md) | Default package app pattern after integration smoke test passes | +| [package-service-pattern.md](./package-service-pattern.md) | General package-service pattern for native long-lived runtimes inside Kody | | [oauth.md](./oauth.md) | **Start here** for third-party OAuth (`/connect/oauth`, redirect URI, params) | | [generated-ui-oauth.md](./generated-ui-oauth.md) | Edge case: OAuth in a hosted package app (`open_generated_ui` on a saved package) | | [connect-secret.md](./connect-secret.md) | `/connect/secret` URL parameters and policies | diff --git a/docs/guides/package-service-pattern.md b/docs/guides/package-service-pattern.md new file mode 100644 index 0000000000..87adfa9428 --- /dev/null +++ b/docs/guides/package-service-pattern.md @@ -0,0 +1,193 @@ +# Package service pattern + +Use this guide when you want a **native Kody package** to own a long-lived +runtime instead of relying on a separate external service. + +This guide assumes: + +- the package runtime already supports `package.json#kody.services` +- the package will likely also expose a package app for operator UI +- protocol credentials and downstream delivery details are handled separately + +## Recommended package shape + +Prefer one saved package with: + +- `package.json#kody.app.entry` for setup UI, dashboards, and callback endpoints +- `package.json#kody.services.realtime-supervisor.entry` for the long-lived + runtime +- package exports for reusable helpers and formatting logic + +Example manifest shape: + +```json +{ + "name": "@scope/realtime-supervisor", + "exports": { + ".": "./src/index.ts", + "./format-event": "./src/format-event.ts" + }, + "kody": { + "id": "realtime-supervisor", + "description": "Native long-lived service package", + "app": { + "entry": "./src/app.ts" + }, + "services": { + "realtime-supervisor": { + "entry": "./src/services/realtime-supervisor.ts", + "autoStart": true, + "timeoutMs": 300000 + } + } + } +} +``` + +## Runtime model + +Package services run as **background-managed** service instances: + +- `service_start` returns immediately with a running state +- the service Durable Object keeps lifecycle state such as `status`, + `active_run_id`, `stop_requested`, and `next_alarm_at` +- the service module runs with: + - `packageContext` + - `serviceContext` + - service-owned writable `storage` + - a `service` helper from `kody:runtime` + +Import shape inside a service module: + +```ts +import { service, serviceContext, storage } from 'kody:runtime' +``` + +The `service` helper exposes: + +- `await service.getStatus()` +- `await service.shouldStop()` +- `await service.setAlarm(runAt)` +- `await service.clearAlarm()` + +## Recommended service loop + +Treat the service entry as a **runtime supervisor**, not just a one-shot job. + +Recommended loop: + +1. Load persisted session state from `storage` + - configuration references + - last checkpoint / cursor / offset + - reconnect endpoint or transport metadata + - topology metadata if applicable +2. Open the outbound connection or initialize the runtime session +3. Authenticate, subscribe, or resume +4. Enter a loop that: + - reads remote events or messages + - persists resumable state before acting on them + - publishes normalized events to the package app or package exports + - periodically checks `await service.shouldStop()` +5. On clean shutdown: + - close the connection + - clear alarms when no reconnect is desired +6. On transient disconnect: + - persist the latest resumable state + - schedule reconnect with `await service.setAlarm(...)` + - exit the current run + +Pseudo-code shape: + +```ts +import { service, storage } from 'kody:runtime' + +async function openSocket(session: unknown) { + void session + return await new Promise((resolve, reject) => { + const socket = new WebSocket('wss://example.com/stream') + socket.addEventListener('open', () => resolve(socket), { once: true }) + socket.addEventListener('error', () => reject(new Error('Failed to open stream')), { + once: true, + }) + }) +} + +export default async function run() { + const session = (await storage.get('session-state')) ?? null + const socket = await openSocket(session) + + try { + // Authenticate, subscribe, or resume here. + + while ( + socket.readyState !== WebSocket.CLOSING && + socket.readyState !== WebSocket.CLOSED + ) { + if (await service.shouldStop()) { + socket.close(1000, 'service stop requested') + await service.clearAlarm() + return { stopped: true } + } + + const event = await readNextEvent(socket) + await persistRuntimeState(storage, event) + await handleRuntimeEvent(event) + } + } catch (error) { + await persistFailure(storage, error) + await service.setAlarm(new Date(Date.now() + 5_000)) + throw error + } +} +``` + +## Use the package app as the operator plane + +Do not push all operational interaction into the service itself. + +Use the package app for: + +- health dashboards +- reconnect / pause / resume controls +- logs +- live event inspection +- setup flows + +The package app can call the package service lifecycle surface and consume +service status through the existing package runtime bridge. + +## Prefer bounded reconnect loops over one immortal run + +Cloudflare Durable Objects can open outbound WebSockets, but outgoing WebSockets +do **not** hibernate. Favor a design where the service can: + +- reconnect intentionally +- reschedule itself with alarms +- persist resumable state early +- stop cooperatively + +Use `timeoutMs` to give the service enough room to run, but do not rely on +indefinite execution as the only lifecycle mechanism. + +## What to persist + +At minimum, persist: + +- reconnect/session identifiers +- latest offset / sequence / cursor +- topology metadata if needed +- reconnect URL or endpoint metadata +- last error / last disconnect reason + +Persist before any expensive or lossy downstream work so reconnects can resume +from durable state. + +## Downstream delivery + +A long-lived package service usually needs a second delivery plane: + +- package app realtime sessions for human/operator UIs +- package exports or jobs for reusable downstream logic + +Keep connection ingest, downstream formatting, and UI concerns separate so the +service can stay focused on connection lifecycle and event normalization. diff --git a/docs/use/execute.md b/docs/use/execute.md index 30e180df24..4b99b49266 100644 --- a/docs/use/execute.md +++ b/docs/use/execute.md @@ -21,6 +21,18 @@ and export a default function. These helpers are runtime exports: active execute or job params instead of receiving them as a function argument - use **`import { packageContext } from 'kody:runtime'`** inside saved package code when you need package metadata; it is **`null`** for ad hoc execute calls +- use **`import { serviceContext } from 'kody:runtime'`** inside package service + code when you need the current service identity; it is **`null`** outside + package service runs +- package service runs also expose **`service`** through **`kody:runtime`** for + background lifecycle control: + - `await service.getStatus()` — read the current package-service status + - `await service.shouldStop()` — cooperatively observe stop requests + - `await service.setAlarm(runAt)` — schedule the next service wake-up + - `await service.clearAlarm()` — clear a pending service wake-up +- package service runs may also declare **`kody.services..timeoutMs`** in + `package.json` when they need a longer executor budget than the default + package-service timeout - use **`import thing from 'kody:@scope/my-package/export-name'`** to reuse a saved package export by full package name @@ -52,6 +64,8 @@ module-oriented runtime model: - package-specific metadata lives under `package.json#kody` - package jobs are schedules declared under `package.json#kody.jobs` - package apps are optional UI surfaces declared under `package.json#kody.app` +- package services are optional long-lived runtimes declared under + `package.json#kody.services` - non-package jobs can also be scheduled directly with **`codemode.job_schedule(...)`** without creating a saved package - **`codemode.job_schedule_once(...)`** remains available as a convenience alias @@ -94,6 +108,11 @@ package-owned jobs and non-package jobs created with `job_schedule` or `job_schedule_once`. - bound storage is execute-, app-, package-, or job-owned durable state +- package service runs also get writable service-owned durable state scoped to + the declared service name +- package service runs are background-managed by the service Durable Object, so + `service_start` returns immediately with a running state while the service code + continues in the background until it finishes, errors, or cooperatively stops - import **`storage`** from **`kody:runtime`** - use **`storage.get(...)`**, **`storage.set(...)`**, **`storage.list(...)`**, and **`storage.sql(query, params?)`** diff --git a/docs/use/packages.md b/docs/use/packages.md index 6e33757943..6418d4acd3 100644 --- a/docs/use/packages.md +++ b/docs/use/packages.md @@ -13,6 +13,7 @@ Think in terms of: - packages - package exports - package apps +- package services - package-owned jobs Packages are the saved-entity unit across search, execute, repo editing, and UI @@ -30,6 +31,7 @@ Important fields: - `kody.description` — package description for search/detail - `kody.tags` — package tags - `kody.app` — optional hosted package app config +- `kody.services` — optional package-owned service runtimes - `kody.jobs` — optional package-owned schedules `package.json` is the manifest. @@ -70,6 +72,34 @@ Treat package apps like Worker-style modules: - internal Durable Objects or facets are implementation details, not the public authoring contract +## Package services + +A package service is optional. + +When `package.json#kody.services` is present, the package can declare one or +more named service entrypoints that Kody runs with package-owned storage and +package caller context. + +Use the package service model when the package needs: + +- long-lived or repeated background work +- package-owned daemon-like logic +- package state that is separate from browser sessions +- a service that should publish updates into a package app + +Treat package services like package-owned runtime modules: + +- service code lives in the package repo +- each service entry module is declared by `kody.services..entry` +- services may optionally declare `kody.services..timeoutMs` to raise the + executor timeout for long-lived or connector-style runs +- service lifecycle is controlled through the `services` capability domain +- service starts return immediately and the service keeps running in the + background until it finishes or is stopped +- service code can inspect its own lifecycle through `serviceContext` and the + `service` helper exposed by `kody:runtime` +- services share the same saved package identity as package apps and jobs + ## Package-owned jobs Packages can own jobs, and Kody also supports schedules that are not owned diff --git a/packages/worker/src/env-schema.ts b/packages/worker/src/env-schema.ts index d9f2b4bf64..eab2fcef11 100644 --- a/packages/worker/src/env-schema.ts +++ b/packages/worker/src/env-schema.ts @@ -208,6 +208,18 @@ export const EnvSchema = object({ context.path, ) }), + PACKAGE_SERVICE_INSTANCE: createSchema< + unknown, + DurableObjectNamespace + >((value, context) => { + if (value) { + return { value: value as DurableObjectNamespace } + } + return fail( + 'Missing PACKAGE_SERVICE_INSTANCE binding for package service runtimes.', + context.path, + ) + }), APP_BASE_URL: optionalUrlStringSchema, APP_COMMIT_SHA: optionalCommitShaSchema, CLOUDFLARE_EMAIL_FROM: optionalNonEmptyStringSchema, diff --git a/packages/worker/src/index.ts b/packages/worker/src/index.ts index 672ab6a37d..bd27a3e638 100644 --- a/packages/worker/src/index.ts +++ b/packages/worker/src/index.ts @@ -9,6 +9,7 @@ import { StorageRunner } from './storage-runner.ts' import { AgentTurnRunner } from './agent-turn/runner-do.ts' import { RepoSession } from './repo/repo-session-do.ts' import { PackageRealtimeSession } from '#worker/package-runtime/realtime-session.ts' +import { PackageServiceInstance } from '#worker/package-runtime/package-service.ts' import { chatAgentBasePath } from '@kody-internal/shared/chat-routes.ts' import { getWorkerSentryOptions } from './sentry-options.ts' import { handleRequest } from '#app/handler.ts' @@ -54,6 +55,7 @@ export { MCP, JobManager, PackageRealtimeSession, + PackageServiceInstance, PackageAppRuntimeBridge, StorageRunner, } diff --git a/packages/worker/src/jobs/service.ts b/packages/worker/src/jobs/service.ts index 55ee9c2a23..fd63c1efc3 100644 --- a/packages/worker/src/jobs/service.ts +++ b/packages/worker/src/jobs/service.ts @@ -185,6 +185,7 @@ async function persistPublishedJobBundleArtifact(input: { modules: bundle.modules, dependencies: bundle.dependencies, packageContext: input.packageContext ?? null, + serviceContext: null, createdAt: new Date().toISOString(), } await persistPublishedBundleArtifact({ diff --git a/packages/worker/src/mcp/capabilities/builtin-domains.ts b/packages/worker/src/mcp/capabilities/builtin-domains.ts index f659504ce8..8382c65090 100644 --- a/packages/worker/src/mcp/capabilities/builtin-domains.ts +++ b/packages/worker/src/mcp/capabilities/builtin-domains.ts @@ -5,6 +5,7 @@ import { metaDomain } from './meta/domain.ts' import { packagesDomain } from './packages/domain.ts' import { repoDomain } from './repo/domain.ts' import { secretsDomain } from './secrets/domain.ts' +import { servicesDomain } from './services/domain.ts' import { storageDomain } from './storage/domain.ts' import { valuesDomain } from './values/domain.ts' @@ -24,6 +25,7 @@ export const builtinDomains = [ packagesDomain, repoDomain, secretsDomain, + servicesDomain, storageDomain, valuesDomain, ] as const diff --git a/packages/worker/src/mcp/capabilities/coding/kody-official-guide.node.test.ts b/packages/worker/src/mcp/capabilities/coding/kody-official-guide.node.test.ts index cc59ffc0c0..29baee25db 100644 --- a/packages/worker/src/mcp/capabilities/coding/kody-official-guide.node.test.ts +++ b/packages/worker/src/mcp/capabilities/coding/kody-official-guide.node.test.ts @@ -32,6 +32,11 @@ test('kody_official_guide returns markdown when fetch succeeds', async () => { } }) +test('kody_official_guide exposes the package service pattern guide', () => { + const url = buildKodyOfficialGuideUrlForTest('package_service_pattern') + expect(url).toMatch(/\/package-service-pattern\.md$/) +}) + test('kody_official_guide surfaces fetch failures', async () => { const originalFetch = globalThis.fetch try { diff --git a/packages/worker/src/mcp/capabilities/coding/kody-official-guide.ts b/packages/worker/src/mcp/capabilities/coding/kody-official-guide.ts index a350d21798..c6529001e2 100644 --- a/packages/worker/src/mcp/capabilities/coding/kody-official-guide.ts +++ b/packages/worker/src/mcp/capabilities/coding/kody-official-guide.ts @@ -52,16 +52,17 @@ export const kodyOfficialGuideCatalog = { summary: 'Hosted /connect/secret URL shape, query params, and approval policy for API keys and PATs.', }, + package_service_pattern: { + file: 'package-service-pattern.md', + title: 'Package service pattern guide', + summary: + 'Build a package-native long-lived service using kody.services, package app realtime fanout, service-owned storage, background lifecycle helpers, and scheduled wake-ups.', + }, } as const -const guideIds = Object.keys(kodyOfficialGuideCatalog) as [ - 'integration_bootstrap', - 'secret_backed_integration', - 'integration_backed_app', - 'oauth', - 'generated_ui_oauth', - 'connect_secret', -] +const guideIds = Object.keys( + kodyOfficialGuideCatalog, +) as Array function buildRawGithubUrl(file: string): string { const { owner, repo, ref, basePath } = KODY_GUIDES_REPO @@ -110,7 +111,7 @@ function buildCapabilityDescription(): string { return `- \`${id}\`: ${g.summary}` }) return [ - 'Load an official Kody guide from the kody GitHub repository (markdown). **For third-party integrations that will power a package, package app, or workflow, use `guide: "integration_bootstrap"` first.** For non-OAuth APIs backed by saved secrets, then use `guide: "secret_backed_integration"` as the default recipe. After the smoke test passes and you are ready to build a package app, use `guide: "integration_backed_app"` for the default package-app pattern. For OAuth mechanics, then use `guide: "oauth"` (standard `/connect/oauth` path). Use `generated_ui_oauth` only for custom package-app OAuth. For API keys/PATs, use `connect_secret` for secret collection. If you are unsure, **call this capability** with the right `guide` instead of guessing.', + 'Load an official Kody guide from the kody GitHub repository (markdown). **For third-party integrations that will power a package, package app, or workflow, use `guide: "integration_bootstrap"` first.** For non-OAuth APIs backed by saved secrets, then use `guide: "secret_backed_integration"` as the default recipe. After the smoke test passes and you are ready to build a package app, use `guide: "integration_backed_app"` for the default package-app pattern. For OAuth mechanics, then use `guide: "oauth"` (standard `/connect/oauth` path). Use `generated_ui_oauth` only for custom package-app OAuth. For API keys/PATs, use `connect_secret` for secret collection. For package-native long-lived service work built on `kody.services`, use `package_service_pattern`. If you are unsure, **call this capability** with the right `guide` instead of guessing.', '', 'Available guides (order matters—start with `integration_bootstrap` for integration-dependent work):', ...lines, @@ -120,7 +121,12 @@ function buildCapabilityDescription(): string { } const guideFieldSchema = z - .enum(guideIds) + .enum( + guideIds as [ + KodyOfficialGuideId, + ...Array, + ], + ) .describe( [ 'Which guide to load.', @@ -130,6 +136,7 @@ const guideFieldSchema = z '`oauth`: standard third-party OAuth via /connect/oauth (read this first for OAuth).', '`generated_ui_oauth`: edge case—OAuth in a hosted package app.', '`connect_secret`: /connect/secret for API keys, PATs, and other secret collection steps.', + '`package_service_pattern`: package-native long-lived service architecture built on package services and package app realtime.', ].join(' '), ) @@ -175,6 +182,14 @@ const allKeywords = [ 'api key', 'personal access token', 'connect secret', + 'package service', + 'service package', + 'background service', + 'long-lived service', + 'outbound websocket', + 'heartbeat', + 'reconnect', + 'resume', 'credentials', 'official guide', 'documentation', diff --git a/packages/worker/src/mcp/capabilities/domain-metadata.ts b/packages/worker/src/mcp/capabilities/domain-metadata.ts index 4b7d68cad7..f9b7d4c2cb 100644 --- a/packages/worker/src/mcp/capabilities/domain-metadata.ts +++ b/packages/worker/src/mcp/capabilities/domain-metadata.ts @@ -9,6 +9,7 @@ export const capabilityDomainNames = { packages: 'packages', repo: 'repo', secrets: 'secrets', + services: 'services', storage: 'storage', values: 'values', } as const diff --git a/packages/worker/src/mcp/capabilities/services/domain.ts b/packages/worker/src/mcp/capabilities/services/domain.ts new file mode 100644 index 0000000000..c4769d1a26 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/services/domain.ts @@ -0,0 +1,19 @@ +import { defineDomain } from '../define-domain.ts' +import { capabilityDomainNames } from '../domain-metadata.ts' +import { serviceGetCapability } from './service-get.ts' +import { serviceListCapability } from './service-list.ts' +import { serviceStartCapability } from './service-start.ts' +import { serviceStopCapability } from './service-stop.ts' + +export const servicesDomain = defineDomain({ + name: capabilityDomainNames.services, + description: + 'Package service capabilities for inspecting and controlling long-lived package-owned runtimes.', + keywords: ['service', 'package service', 'runtime', 'background', 'gateway'], + capabilities: [ + serviceListCapability, + serviceGetCapability, + serviceStartCapability, + serviceStopCapability, + ], +}) diff --git a/packages/worker/src/mcp/capabilities/services/service-get.ts b/packages/worker/src/mcp/capabilities/services/service-get.ts new file mode 100644 index 0000000000..b61a7b10e9 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/services/service-get.ts @@ -0,0 +1,42 @@ +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { z } from 'zod' +import { + normalizePackageServiceStatus, + packageServiceStatusSchema, + requirePackageServiceContext, +} from './shared.ts' + +export const serviceGetCapability = defineDomainCapability( + capabilityDomainNames.services, + { + name: 'service_get', + description: + 'Get status for one package service declared under package.json#kody.services.', + keywords: ['service', 'status', 'package', 'inspect'], + readOnly: true, + idempotent: true, + destructive: false, + inputSchema: z.object({ + service_name: z.string().trim().min(1), + package_id: z.string().min(1).optional(), + }), + outputSchema: packageServiceStatusSchema, + async handler(args, ctx) { + const serviceContext = await requirePackageServiceContext({ + env: ctx.env, + callerContext: ctx.callerContext, + serviceName: args.service_name, + explicitPackageId: args.package_id, + }) + if (!serviceContext.service) { + throw new Error( + `Package service "${args.service_name}" was not found.`, + ) + } + return normalizePackageServiceStatus( + await serviceContext.service.status(), + ) + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/services/service-list.ts b/packages/worker/src/mcp/capabilities/services/service-list.ts new file mode 100644 index 0000000000..e6868cfbb5 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/services/service-list.ts @@ -0,0 +1,65 @@ +import { z } from 'zod' +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { + listPackageServicesForContext, + packageServiceSummarySchema, + normalizePackageServiceStatus, +} from './shared.ts' + +const inputSchema = z.object({ + package_id: z.string().min(1).optional(), +}) + +const outputSchema = z.object({ + package_id: z.string(), + kody_id: z.string(), + services: z.array(packageServiceSummarySchema), +}) + +export const serviceListCapability = defineDomainCapability( + capabilityDomainNames.services, + { + name: 'service_list', + description: + 'List package services declared by a saved package, including their current runtime status.', + keywords: ['service', 'services', 'package', 'list', 'status'], + readOnly: true, + idempotent: true, + destructive: false, + inputSchema, + outputSchema, + async handler(args, ctx) { + const listed = await listPackageServicesForContext({ + env: ctx.env, + callerContext: ctx.callerContext, + explicitPackageId: args.package_id, + }) + const services = await Promise.all( + listed.services.map(async (service) => { + let status: z.infer['status'] = + 'unknown' + try { + status = normalizePackageServiceStatus( + await listed.rpc(service.name).status(), + ).status + } catch { + // Keep the rest of the service list usable if one status lookup fails. + } + return { + name: service.name, + entry: service.entry, + auto_start: service.auto_start, + timeout_ms: service.timeout_ms ?? null, + status, + } + }), + ) + return { + package_id: listed.savedPackage.id, + kody_id: listed.savedPackage.kodyId, + services, + } + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/services/service-start.ts b/packages/worker/src/mcp/capabilities/services/service-start.ts new file mode 100644 index 0000000000..9a24f893f8 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/services/service-start.ts @@ -0,0 +1,46 @@ +import { z } from 'zod' +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { requirePackageServiceContext } from './shared.ts' + +const inputSchema = z.object({ + service_name: z.string().trim().min(1), + package_id: z.string().min(1).optional(), +}) + +const outputSchema = z.object({ + ok: z.boolean(), + run_id: z.string().optional(), + started_at: z.string().optional(), + status: z.enum(['running']).optional(), + already_running: z.boolean().optional(), +}) + +export const serviceStartCapability = defineDomainCapability( + capabilityDomainNames.services, + { + name: 'service_start', + description: + 'Start a package service instance and return the latest execution result.', + keywords: ['service', 'start', 'package service', 'runtime'], + readOnly: false, + idempotent: false, + destructive: false, + inputSchema, + outputSchema, + async handler(args, ctx) { + const serviceContext = await requirePackageServiceContext({ + env: ctx.env, + callerContext: ctx.callerContext, + serviceName: args.service_name, + explicitPackageId: args.package_id, + }) + if (!serviceContext.service) { + throw new Error( + `Package service "${args.service_name}" was not found for this package.`, + ) + } + return await serviceContext.service.start() + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/services/service-stop.ts b/packages/worker/src/mcp/capabilities/services/service-stop.ts new file mode 100644 index 0000000000..66e7feebf3 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/services/service-stop.ts @@ -0,0 +1,44 @@ +import { z } from 'zod' +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { requirePackageServiceContext } from './shared.ts' + +const inputSchema = z.object({ + service_name: z.string().trim().min(1), + package_id: z.string().min(1).optional(), +}) + +const outputSchema = z.object({ + ok: z.boolean(), +}) + +export const serviceStopCapability = defineDomainCapability( + capabilityDomainNames.services, + { + name: 'service_stop', + description: 'Stop one declared package service instance.', + keywords: ['service', 'stop', 'package'], + readOnly: false, + idempotent: false, + destructive: false, + inputSchema, + outputSchema, + async handler(args, ctx) { + const serviceContext = await requirePackageServiceContext({ + env: ctx.env, + callerContext: ctx.callerContext, + serviceName: args.service_name, + explicitPackageId: args.package_id, + }) + if (!serviceContext.service) { + throw new Error( + `Package service "${args.service_name}" was not found.`, + ) + } + const result = (await serviceContext.service.stop()) as { ok?: unknown } + return { + ok: result?.ok === true, + } + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/services/services-domain.node.test.ts b/packages/worker/src/mcp/capabilities/services/services-domain.node.test.ts new file mode 100644 index 0000000000..6128b5caeb --- /dev/null +++ b/packages/worker/src/mcp/capabilities/services/services-domain.node.test.ts @@ -0,0 +1,319 @@ +import { expect, test, vi } from 'vitest' +import { createMcpCallerContext } from '#mcp/context.ts' + +const mockModule = vi.hoisted(() => ({ + listSavedPackageServices: vi.fn(), + packageServiceRpc: vi.fn(), + getSavedPackageById: vi.fn(), + normalizePackageServiceStatus: vi.fn((input: unknown) => input), + packageServiceStatusSchema: { + parse: vi.fn((input: unknown) => input), + safeParse: vi.fn((input: unknown) => ({ + success: true, + data: input, + })), + }, +})) + +vi.mock('#worker/package-runtime/package-service.ts', () => ({ + listSavedPackageServices: (...args: Array) => + mockModule.listSavedPackageServices(...args), + packageServiceRpc: (...args: Array) => + mockModule.packageServiceRpc(...args), + normalizePackageServiceStatus: (...args: Array) => + mockModule.normalizePackageServiceStatus(...args), + packageServiceStatusSchema: mockModule.packageServiceStatusSchema, +})) + +vi.mock('#worker/package-registry/repo.ts', () => ({ + getSavedPackageById: (...args: Array) => + mockModule.getSavedPackageById(...args), +})) + +const { servicesDomain } = await import('./domain.ts') +const { serviceListCapability } = await import('./service-list.ts') +const { serviceGetCapability } = await import('./service-get.ts') +const { serviceStartCapability } = await import('./service-start.ts') +const { serviceStopCapability } = await import('./service-stop.ts') + +function resetMocks() { + mockModule.listSavedPackageServices.mockReset() + mockModule.packageServiceRpc.mockReset() + mockModule.getSavedPackageById.mockReset() + mockModule.normalizePackageServiceStatus.mockClear() + mockModule.packageServiceStatusSchema.parse.mockClear() + mockModule.packageServiceStatusSchema.safeParse.mockClear() +} + +function createCallerContext() { + return createMcpCallerContext({ + baseUrl: 'https://example.com', + user: { + userId: 'user-123', + email: 'user@example.com', + displayName: 'User Example', + }, + storageContext: { + sessionId: null, + appId: 'package-123', + storageId: 'package-123', + }, + }) +} + +test('services domain exposes package service lifecycle capabilities', () => { + expect(servicesDomain.capabilities.map((capability) => capability.name)).toEqual( + expect.arrayContaining([ + 'service_list', + 'service_get', + 'service_start', + 'service_stop', + ]), + ) +}) + +test('service_list returns declared package services with live status', async () => { + resetMocks() + mockModule.getSavedPackageById.mockResolvedValue({ + id: 'package-123', + userId: 'user-123', + name: '@scope/example', + kodyId: 'example', + description: 'Example package', + tags: [], + searchText: null, + sourceId: 'source-123', + hasApp: true, + createdAt: '2026-04-24T00:00:00.000Z', + updatedAt: '2026-04-24T00:00:00.000Z', + }) + mockModule.listSavedPackageServices.mockResolvedValue({ + savedPackage: { + id: 'package-123', + kodyId: 'example', + }, + services: [ + { + name: 'realtime-supervisor', + entry: 'services/realtime-supervisor.ts', + autoStart: true, + timeoutMs: 300000, + }, + ], + rpc: () => ({ + status: async () => ({ + package_id: 'package-123', + kody_id: 'example', + service_name: 'realtime-supervisor', + status: 'stopped', + auto_start: true, + timeout_ms: 300000, + stop_requested: false, + active_run_id: null, + next_alarm_at: null, + last_error: null, + last_started_at: null, + last_stopped_at: null, + last_run_finished_at: null, + last_result: null, + }), + }), + }) + mockModule.packageServiceRpc.mockImplementation(() => ({ + status: async () => ({ + package_id: 'package-123', + kody_id: 'example', + service_name: 'realtime-supervisor', + status: 'stopped', + auto_start: true, + timeout_ms: 300000, + stop_requested: false, + active_run_id: null, + next_alarm_at: null, + last_error: null, + last_started_at: null, + last_stopped_at: null, + last_run_finished_at: null, + last_result: null, + }), + })) + const result = await serviceListCapability.handler( + {}, + { + env: { + APP_DB: {} as D1Database, + } as Env, + callerContext: createCallerContext(), + }, + ) + + expect(result).toEqual({ + package_id: 'package-123', + kody_id: 'example', + services: [ + { + name: 'realtime-supervisor', + entry: 'services/realtime-supervisor.ts', + auto_start: true, + status: 'stopped', + timeout_ms: 300000, + }, + ], + }) +}) + +test('service_list marks status as unknown when a service status lookup fails', async () => { + resetMocks() + mockModule.getSavedPackageById.mockResolvedValue({ + id: 'package-123', + userId: 'user-123', + name: '@scope/example', + kodyId: 'example', + description: 'Example package', + tags: [], + searchText: null, + sourceId: 'source-123', + hasApp: true, + createdAt: '2026-04-24T00:00:00.000Z', + updatedAt: '2026-04-24T00:00:00.000Z', + }) + mockModule.listSavedPackageServices.mockResolvedValue({ + savedPackage: { + id: 'package-123', + kodyId: 'example', + }, + services: [ + { + name: 'realtime-supervisor', + entry: 'services/realtime-supervisor.ts', + autoStart: false, + timeoutMs: null, + }, + ], + rpc: () => ({ + status: async () => { + throw new Error('worker unavailable') + }, + }), + }) + + const result = await serviceListCapability.handler( + {}, + { + env: { + APP_DB: {} as D1Database, + } as Env, + callerContext: createCallerContext(), + }, + ) + + expect(result).toEqual({ + package_id: 'package-123', + kody_id: 'example', + services: [ + { + name: 'realtime-supervisor', + entry: 'services/realtime-supervisor.ts', + auto_start: false, + status: 'unknown', + timeout_ms: null, + }, + ], + }) +}) + +test('service_get, service_start, and service_stop delegate to package service RPC', async () => { + resetMocks() + mockModule.getSavedPackageById.mockResolvedValue({ + id: 'package-123', + userId: 'user-123', + name: '@scope/example', + kodyId: 'example', + description: 'Example package', + tags: [], + searchText: null, + sourceId: 'source-123', + hasApp: true, + createdAt: '2026-04-24T00:00:00.000Z', + updatedAt: '2026-04-24T00:00:00.000Z', + }) + mockModule.packageServiceRpc.mockImplementation(() => ({ + status: async () => ({ + package_id: 'package-123', + kody_id: 'example', + service_name: 'realtime-supervisor', + status: 'running', + auto_start: false, + timeout_ms: 300000, + stop_requested: false, + active_run_id: 'run-123', + next_alarm_at: null, + last_error: null, + last_started_at: null, + last_stopped_at: null, + last_run_finished_at: null, + last_result: null, + }), + start: async () => ({ + ok: true, + run_id: 'run-123', + started_at: '2026-04-24T00:00:00.000Z', + status: 'running', + }), + stop: async () => ({ + ok: true, + }), + })) + + const env = { + APP_DB: {} as D1Database, + } as Env + const callerContext = createCallerContext() + + await expect( + serviceGetCapability.handler( + { + service_name: 'realtime-supervisor', + }, + { + env, + callerContext, + }, + ), + ).resolves.toMatchObject({ + service_name: 'realtime-supervisor', + status: 'running', + active_run_id: 'run-123', + }) + + await expect( + serviceStartCapability.handler( + { + service_name: 'realtime-supervisor', + }, + { + env, + callerContext, + }, + ), + ).resolves.toEqual({ + ok: true, + run_id: 'run-123', + started_at: '2026-04-24T00:00:00.000Z', + status: 'running', + }) + + await expect( + serviceStopCapability.handler( + { + service_name: 'realtime-supervisor', + }, + { + env, + callerContext, + }, + ), + ).resolves.toEqual({ + ok: true, + }) +}) diff --git a/packages/worker/src/mcp/capabilities/services/shared.ts b/packages/worker/src/mcp/capabilities/services/shared.ts new file mode 100644 index 0000000000..ddacc5117f --- /dev/null +++ b/packages/worker/src/mcp/capabilities/services/shared.ts @@ -0,0 +1,122 @@ +import { type McpCallerContext } from '@kody-internal/shared/chat.ts' +import { z } from 'zod' +import { requireMcpUser } from '#mcp/capabilities/meta/require-user.ts' +import { getSavedPackageById } from '#worker/package-registry/repo.ts' +import { + listSavedPackageServices, + normalizePackageServiceStatus, + packageServiceRpc, + packageServiceStatusSchema, +} from '#worker/package-runtime/package-service.ts' + +export { normalizePackageServiceStatus, packageServiceStatusSchema } + +export const packageServiceRecordSchema = z.object({ + name: z.string(), + entry: z.string(), + auto_start: z.boolean(), + timeout_ms: z.number().int().positive().nullable(), +}) + +export const packageServiceSummarySchema = packageServiceRecordSchema.extend({ + status: z.enum(['idle', 'running', 'stopping', 'stopped', 'error', 'unknown']), +}) + +function resolvePackageId( + callerContext: McpCallerContext, + explicitPackageId?: string | null, +) { + const normalizedExplicit = + typeof explicitPackageId === 'string' ? explicitPackageId.trim() : '' + if (normalizedExplicit) { + return normalizedExplicit + } + const appId = callerContext.storageContext?.appId?.trim() || '' + if (appId) { + return appId + } + throw new Error( + 'Package service APIs require package app or package job caller context.', + ) +} + +export async function requirePackageServiceContext(input: { + env: Env + callerContext: McpCallerContext + explicitPackageId?: string | null + serviceName?: string | null +}) { + const user = requireMcpUser(input.callerContext) + const packageId = resolvePackageId( + input.callerContext, + input.explicitPackageId, + ) + const savedPackage = await getSavedPackageById(input.env.APP_DB, { + userId: user.userId, + packageId, + }) + if (!savedPackage) { + throw new Error('Saved package was not found for package service operations.') + } + const serviceName = input.serviceName?.trim() || '' + return { + user, + savedPackage, + serviceName, + service: serviceName + ? packageServiceRpc({ + env: input.env, + userId: user.userId, + packageId: savedPackage.id, + kodyId: savedPackage.kodyId, + sourceId: savedPackage.sourceId, + baseUrl: input.callerContext.baseUrl, + serviceName, + }) + : null, + } +} + +export async function listPackageServicesForContext(input: { + env: Env + callerContext: McpCallerContext + explicitPackageId?: string | null +}) { + const context = await requirePackageServiceContext({ + env: input.env, + callerContext: input.callerContext, + explicitPackageId: input.explicitPackageId, + }) + const listed = await listSavedPackageServices({ + env: input.env, + userId: context.user.userId, + baseUrl: input.callerContext.baseUrl, + packageId: context.savedPackage.id, + savedPackage: context.savedPackage, + }) + return { + savedPackage: context.savedPackage, + rpc(serviceName: string) { + return packageServiceRpc({ + env: input.env, + userId: context.user.userId, + packageId: context.savedPackage.id, + kodyId: context.savedPackage.kodyId, + sourceId: context.savedPackage.sourceId, + baseUrl: input.callerContext.baseUrl, + serviceName, + }) + }, + services: listed.services.map((service) => ({ + name: service.name, + entry: service.entry, + auto_start: service.autoStart, + timeout_ms: service.timeoutMs ?? null, + })), + } +} + +export type PackageServiceStatusRecord = z.infer< + typeof packageServiceStatusSchema +> + diff --git a/packages/worker/src/mcp/executor.ts b/packages/worker/src/mcp/executor.ts index 140b95d453..eb5cd82a16 100644 --- a/packages/worker/src/mcp/executor.ts +++ b/packages/worker/src/mcp/executor.ts @@ -23,6 +23,7 @@ export function createExecuteExecutor(input: { exports?: WorkerLoopbackExports gatewayProps: FetchGatewayProps modules?: WorkerLoaderModules + timeoutMs?: number }) { const loopbackExports = input.exports ?? workerExports if (!loopbackExports?.CodemodeFetchGateway) { @@ -32,7 +33,7 @@ export function createExecuteExecutor(input: { } return new DynamicWorkerExecutor({ loader: input.env.LOADER, - timeout: 90_000, + timeout: input.timeoutMs ?? 90_000, globalOutbound: loopbackExports.CodemodeFetchGateway({ props: input.gatewayProps, }), diff --git a/packages/worker/src/mcp/run-codemode-registry.node.test.ts b/packages/worker/src/mcp/run-codemode-registry.node.test.ts index e2b9e2eade..ebfed18afb 100644 --- a/packages/worker/src/mcp/run-codemode-registry.node.test.ts +++ b/packages/worker/src/mcp/run-codemode-registry.node.test.ts @@ -5,6 +5,7 @@ import { buildKodyModuleBundle } from '#worker/package-runtime/module-graph.ts' import { buildCodemodeFns, runCodemodeWithRegistry, + runBundledModuleWithRegistry, runModuleWithRegistry, } from './run-codemode-registry.ts' import * as secretService from '#mcp/secrets/service.ts' @@ -900,3 +901,77 @@ test('runCodemodeWithRegistry keeps legacy snippet execution for non-module code getRegistrySpy.mockRestore() } }) + +test('runBundledModuleWithRegistry injects service helpers and custom timeout', async () => { + const env = {} as Env + const callerContext = createMcpCallerContext({ + baseUrl: 'https://heykody.dev', + user: { userId: 'user-123' }, + }) + const getRegistrySpy = vi + .spyOn( + await import('#mcp/capabilities/registry.ts'), + 'getCapabilityRegistryForContext', + ) + .mockResolvedValue({ + capabilityDomains: [], + capabilityDomainDescriptionsByName: {} as Record, + capabilityHandlers: {}, + capabilityList: [], + capabilityMap: {}, + capabilitySpecs: {}, + capabilityToolDescriptors: {}, + } as Awaited>) + let wrapped = '' + const createExecuteExecutorSpy = vi + .spyOn(await import('#mcp/executor.ts'), 'createExecuteExecutor') + .mockImplementation((input) => { + expect(input.timeoutMs).toBe(300_000) + return { + async execute(source) { + wrapped = String(source) + return { + result: 'ok', + logs: [], + } + }, + } as never + }) + + try { + const result = await runBundledModuleWithRegistry( + env, + callerContext, + { + mainModule: 'entry.js', + modules: { + 'entry.js': 'export default async () => "ok"', + }, + }, + undefined, + { + serviceContext: { + serviceName: 'realtime-supervisor', + }, + serviceTools: { + getStatus: async () => ({ status: 'running' }), + shouldStop: async () => false, + setAlarm: async () => ({ ok: true, scheduled_at: '2026-04-25T00:00:00.000Z' }), + clearAlarm: async () => ({ ok: true }), + }, + executorTimeoutMs: 300_000, + }, + ) + + expect(result.result).toBe('ok') + expect(wrapped).toContain('const service = {') + expect(wrapped).toContain("service_get_status") + expect(wrapped).toContain("service_should_stop") + expect(wrapped).toContain("service_set_alarm") + expect(wrapped).toContain("service_clear_alarm") + expect(wrapped).toContain('"serviceName":"realtime-supervisor"') + } finally { + createExecuteExecutorSpy.mockRestore() + getRegistrySpy.mockRestore() + } +}) diff --git a/packages/worker/src/mcp/run-codemode-registry.ts b/packages/worker/src/mcp/run-codemode-registry.ts index fd65d0ecb9..20d5e71efa 100644 --- a/packages/worker/src/mcp/run-codemode-registry.ts +++ b/packages/worker/src/mcp/run-codemode-registry.ts @@ -48,6 +48,31 @@ type StorageToolOptions = { writable: boolean } +type ServiceToolOptions = { + getStatus: () => Promise + shouldStop: () => Promise + setAlarm: (runAt: Date) => Promise<{ ok: true; scheduled_at: string }> + clearAlarm: () => Promise<{ ok: true }> +} + +function createServiceHelperPrelude() { + return ` +const service = { + getStatus: async () => await codemode.service_get_status({}), + shouldStop: async () => { + const result = await codemode.service_should_stop({}); + return result?.shouldStop === true; + }, + setAlarm: async (runAt) => { + const normalizedRunAt = + runAt instanceof Date ? runAt.toISOString() : String(runAt ?? ''); + return await codemode.service_set_alarm({ runAt: normalizedRunAt }); + }, + clearAlarm: async () => await codemode.service_clear_alarm({}), +}; + `.trim() +} + export async function buildCodemodeFns( env: Env, callerContext: McpCallerContext, @@ -59,6 +84,7 @@ export async function buildCodemodeFns( trackSecretInputValue?: (value: string) => void additionalTools?: AdditionalCodemodeTools storageTools?: StorageToolOptions + serviceTools?: ServiceToolOptions }, ) { const { capabilityMap } = await getCapabilityRegistryForContext({ @@ -106,9 +132,40 @@ export async function buildCodemodeFns( }) : {} assertNoCapabilityCollisions(capabilityMap, storageCodemodeTools) + const serviceTools = options?.serviceTools + const serviceCodemodeTools: AdditionalCodemodeTools = serviceTools + ? { + service_get_status: async () => + await serviceTools.getStatus(), + service_should_stop: async () => ({ + shouldStop: await serviceTools.shouldStop(), + }), + service_set_alarm: async (args: unknown) => { + const payload = + typeof args === 'object' && args !== null + ? (args as { runAt?: unknown }) + : {} + const runAtValue = payload.runAt + const runAtString = + typeof runAtValue === 'string' ? runAtValue.trim() : '' + if (!runAtString) { + throw new Error('service.setAlarm requires a runAt ISO string.') + } + const runAt = new Date(runAtString) + if (Number.isNaN(runAt.getTime())) { + throw new Error('service.setAlarm requires a valid runAt ISO string.') + } + return await serviceTools.setAlarm(runAt) + }, + service_clear_alarm: async () => + await serviceTools.clearAlarm(), + } + : {} + assertNoCapabilityCollisions(capabilityMap, serviceCodemodeTools) return { ...capabilityCodemodeTools, ...storageCodemodeTools, + ...serviceCodemodeTools, ...additionalTools, } } @@ -131,6 +188,7 @@ export async function buildCodemodeProvider( trackSecretInputValue?: (value: string) => void additionalTools?: AdditionalCodemodeTools storageTools?: StorageToolOptions + serviceTools?: ServiceToolOptions }, ): Promise { const tools = await buildCodemodeFns(env, callerContext, options) @@ -201,7 +259,9 @@ export async function runCodemodeWithRegistry( additionalTools?: AdditionalCodemodeTools helperPrelude?: string storageTools?: StorageToolOptions + serviceTools?: ServiceToolOptions executorModules?: WorkerLoaderModules + executorTimeoutMs?: number }, ): Promise { const moduleSource = stripCodeFences(code.trim()) @@ -210,6 +270,8 @@ export async function runCodemodeWithRegistry( executorExports: options?.executorExports, additionalTools: options?.additionalTools, storageTools: options?.storageTools, + serviceTools: options?.serviceTools, + executorTimeoutMs: options?.executorTimeoutMs, }) } const { createExecuteExecutor } = await import('#mcp/executor.ts') @@ -221,6 +283,7 @@ export async function runCodemodeWithRegistry( const executor = createExecuteExecutor({ env, exports: options?.executorExports ?? workerExports, + timeoutMs: options?.executorTimeoutMs, gatewayProps: { baseUrl: callerContext.baseUrl, userId: callerContext.user?.userId ?? null, @@ -234,6 +297,7 @@ export async function runCodemodeWithRegistry( }, additionalTools: options?.additionalTools, storageTools: options?.storageTools, + serviceTools: options?.serviceTools, }) const wrappedCode = params !== undefined @@ -246,7 +310,14 @@ export async function runCodemodeWithRegistry( writable: options.storageTools.writable, }) : '' - const helperPrelude = [storageHelperPrelude, options?.helperPrelude ?? ''] + const serviceHelperPrelude = options?.serviceTools + ? createServiceHelperPrelude() + : '' + const helperPrelude = [ + storageHelperPrelude, + serviceHelperPrelude, + options?.helperPrelude ?? '', + ] .filter((value) => value.trim().length > 0) .join('\n') const wrapped = `async () => { @@ -279,6 +350,8 @@ export async function runModuleWithRegistry( executorExports?: typeof workerExports additionalTools?: AdditionalCodemodeTools storageTools?: StorageToolOptions + serviceTools?: ServiceToolOptions + executorTimeoutMs?: number }, ): Promise { const userId = callerContext.user?.userId ?? '' @@ -320,6 +393,11 @@ export async function runBundledModuleWithRegistry( packageId: string kodyId: string } | null + serviceContext?: { + serviceName: string + } | null + serviceTools?: ServiceToolOptions + executorTimeoutMs?: number }, ): Promise { const { createExecuteExecutor } = await import('#mcp/executor.ts') @@ -330,6 +408,7 @@ export async function runBundledModuleWithRegistry( const executor = createExecuteExecutor({ env, exports: options?.executorExports ?? workerExports, + timeoutMs: options?.executorTimeoutMs, gatewayProps: { baseUrl: callerContext.baseUrl, userId: callerContext.user?.userId ?? null, @@ -343,6 +422,7 @@ export async function runBundledModuleWithRegistry( }, additionalTools: options?.additionalTools, storageTools: options?.storageTools, + serviceTools: options?.serviceTools, }) const storageHelperPrelude = options?.storageTools ? createStorageHelperPrelude({ @@ -350,9 +430,13 @@ export async function runBundledModuleWithRegistry( writable: options.storageTools.writable, }) : '' + const serviceHelperPrelude = options?.serviceTools + ? createServiceHelperPrelude() + : '' const wrapped = `async () => { ${createExecuteHelperPrelude()} ${storageHelperPrelude ? `${storageHelperPrelude}\n` : ''} +${serviceHelperPrelude ? `${serviceHelperPrelude}\n` : ''} const __previousRuntime = globalThis.__kodyRuntime; globalThis.__kodyRuntime = { codemode, @@ -362,6 +446,8 @@ ${storageHelperPrelude ? `${storageHelperPrelude}\n` : ''} createAuthenticatedFetch, agentChatTurnStream, packageContext: ${JSON.stringify(options?.packageContext ?? null)}, + serviceContext: ${JSON.stringify(options?.serviceContext ?? null)}, + service: typeof service === 'undefined' ? null : service, }; try { const __kodyModule = await import(${JSON.stringify(`./${bundle.mainModule}`)}); diff --git a/packages/worker/src/mcp/server-instructions.ts b/packages/worker/src/mcp/server-instructions.ts index 25c686ab67..16f847bb4b 100644 --- a/packages/worker/src/mcp/server-instructions.ts +++ b/packages/worker/src/mcp/server-instructions.ts @@ -27,8 +27,9 @@ Conventions - \`job_schedule\`: schedule a repo-backed job for the signed-in user without creating a saved package first. Supports one-off, interval, and cron schedules. - \`job_schedule_once\`: compatibility wrapper for one-off repo-backed jobs when you only need a single run time. - \`job_run_now\`: run an existing scheduled job immediately by id and return the updated job view plus execution result for debugging. -- Package jobs are schedules owned by a package. For ad hoc work that is not tied to a package, use \`job_schedule\`. Package apps are optional UI surfaces declared by the package, not a separate top-level primitive. +- Package jobs are schedules owned by a package. For ad hoc work that is not tied to a package, use \`job_schedule\`. Package apps are optional UI surfaces declared by the package, and packages may also declare headless package services for long-lived background runtimes. - Package apps can also use Kody-managed realtime websocket sessions through \`session_list\`, \`session_emit\`, and \`session_broadcast\`, and package jobs can emit to those sessions when running under the same package caller context. +- Package services are inspected and controlled through \`service_list\`, \`service_get\`, \`service_start\`, and \`service_stop\`. - Memory writes are verify-first: always run \`meta_memory_verify\` before \`meta_memory_upsert\` or \`meta_memory_delete\`. Kody retrieves related memories; the consuming agent decides whether to upsert, delete, both, or do nothing. \`meta_memory_upsert\` creates a new memory when \`memory_id\` is omitted and updates an existing memory when \`memory_id\` is provided. - User-specific MCP instructions: \`meta_get_mcp_server_instructions\` / \`meta_set_mcp_server_instructions\` (signed-in users). Updates apply to **new** MCP sessions (reconnect to refresh what the host shows). diff --git a/packages/worker/src/mcp/tools/search.node.test.ts b/packages/worker/src/mcp/tools/search.node.test.ts index 9bf08fa616..677656219b 100644 --- a/packages/worker/src/mcp/tools/search.node.test.ts +++ b/packages/worker/src/mcp/tools/search.node.test.ts @@ -59,6 +59,7 @@ test('searchUnified ranks mixed search rows through one shared pipeline', () => appEntry: null, exports: [], jobs: [], + services: [], }, }, ] @@ -227,6 +228,7 @@ test('optional search rows include saved packages when lookup succeeds', async ( appEntry: 'src/app.ts', exports: ['.'], jobs: [], + services: [], }, }, ], @@ -269,6 +271,7 @@ test('optional search rows preserve package fallback warnings', async () => { appEntry: null, exports: [], jobs: [], + services: [], }, }, ], @@ -318,6 +321,7 @@ test('searchUnified uses package exports and connector aliases for operate queri './playback-controller', ], jobs: [], + services: [], }, }, ], @@ -399,6 +403,7 @@ test('buildSavedPackageSearchRows falls back when package source resolution fail appEntry: null, exports: [], jobs: [], + services: [], }), }), ]) @@ -438,6 +443,7 @@ test('search guidance does not pair unrelated package and connector matches', () appEntry: 'src/app.ts', exports: ['./play'], jobs: [], + services: [], }, }, ], diff --git a/packages/worker/src/mcp/tools/search.ts b/packages/worker/src/mcp/tools/search.ts index 6fd5dd1d8b..bb5310f2bf 100644 --- a/packages/worker/src/mcp/tools/search.ts +++ b/packages/worker/src/mcp/tools/search.ts @@ -170,6 +170,7 @@ function buildFallbackPackageSearchProjection( appEntry: null, exports: [], jobs: [], + services: [], } } diff --git a/packages/worker/src/oauth-handlers.workers.test.ts b/packages/worker/src/oauth-handlers.workers.test.ts index ed835e15b7..80897ef862 100644 --- a/packages/worker/src/oauth-handlers.workers.test.ts +++ b/packages/worker/src/oauth-handlers.workers.test.ts @@ -110,6 +110,7 @@ function createEnv( JOB_MANAGER: mockJobDoNamespace('job-manager-test-id'), STORAGE_RUNNER: mockJobDoNamespace('storage-runner-test-id'), PACKAGE_REALTIME_SESSION: mockJobDoNamespace('package-realtime-session-test-id'), + PACKAGE_SERVICE_INSTANCE: mockJobDoNamespace('package-service-instance-test-id'), } as unknown as Env } diff --git a/packages/worker/src/package-registry/manifest.node.test.ts b/packages/worker/src/package-registry/manifest.node.test.ts index 422afaf9aa..4bab2feef4 100644 --- a/packages/worker/src/package-registry/manifest.node.test.ts +++ b/packages/worker/src/package-registry/manifest.node.test.ts @@ -59,3 +59,58 @@ test('parseAuthoredPackageJson rejects unscoped package names', () => { 'package.json name "cursor-cloud-agents" must be a scoped package name like "@scope/cursor-cloud-agents".', ) }) + +test('parseAuthoredPackageJson accepts package service definitions', () => { + const manifest = parseAuthoredPackageJson({ + content: JSON.stringify({ + name: '@kentcdodds/realtime-supervisor', + exports: { + '.': './index.ts', + }, + kody: { + id: 'realtime-supervisor', + description: 'Realtime supervisor package', + services: { + 'realtime-supervisor': { + entry: './services/realtime-supervisor.ts', + autoStart: true, + timeoutMs: 300000, + }, + }, + }, + }), + manifestPath: 'package.json', + }) + + expect(manifest.kody.services).toEqual({ + 'realtime-supervisor': { + entry: './services/realtime-supervisor.ts', + autoStart: true, + timeoutMs: 300000, + }, + }) +}) + +test('parseAuthoredPackageJson rejects service timeoutMs values above the supported maximum', () => { + expect(() => + parseAuthoredPackageJson({ + content: JSON.stringify({ + name: '@kentcdodds/realtime-supervisor', + exports: { + '.': './index.ts', + }, + kody: { + id: 'realtime-supervisor', + description: 'Realtime supervisor package', + services: { + 'realtime-supervisor': { + entry: './services/realtime-supervisor.ts', + timeoutMs: 300001, + }, + }, + }, + }), + manifestPath: 'package.json', + }), + ).toThrow('expected number to be <=300000') +}) diff --git a/packages/worker/src/package-registry/manifest.ts b/packages/worker/src/package-registry/manifest.ts index 6c76f9e889..8d0e704bb0 100644 --- a/packages/worker/src/package-registry/manifest.ts +++ b/packages/worker/src/package-registry/manifest.ts @@ -117,6 +117,27 @@ export function getPackageAppEntryPath(manifest: AuthoredPackageJson) { return normalizePackageWorkspacePath(appEntry) } +export function getPackageServiceEntryPath(input: { + manifest: AuthoredPackageJson + serviceName: string +}) { + const serviceEntry = + input.manifest.kody.services?.[input.serviceName]?.entry?.trim() + if (!serviceEntry) return null + return normalizePackageWorkspacePath(serviceEntry) +} + +export function listPackageServices(manifest: AuthoredPackageJson) { + return Object.entries(manifest.kody.services ?? {}) + .map(([name, service]) => ({ + name, + entry: normalizePackageWorkspacePath(service.entry), + autoStart: service.autoStart ?? false, + timeoutMs: service.timeoutMs ?? null, + })) + .sort((left, right) => left.name.localeCompare(right.name)) +} + export function getPackageTags(manifest: AuthoredPackageJson) { return [...(manifest.kody.tags ?? [])] } @@ -136,6 +157,12 @@ export type PackageSearchProjection = { schedule: string enabled: boolean }> + services: Array<{ + name: string + entry: string + autoStart: boolean + timeoutMs: number | null + }> } export function buildPackageSearchProjection( @@ -164,6 +191,7 @@ export function buildPackageSearchProjection( enabled: job.enabled ?? true, })) .sort((left, right) => left.name.localeCompare(right.name)), + services: listPackageServices(manifest), } } @@ -173,6 +201,16 @@ export function buildPackageSearchDocument(projection: PackageSearchProjection) .filter((value) => value.length > 0) .join(' '), ) + const serviceLines = projection.services.map((service) => + [ + service.name, + service.entry, + service.autoStart ? 'auto-start' : 'manual-start', + service.timeoutMs != null ? `timeout-ms:${service.timeoutMs}` : '', + ] + .filter((value) => value.length > 0) + .join(' '), + ) return [ `package ${projection.kodyId}`, projection.name, @@ -181,6 +219,7 @@ export function buildPackageSearchDocument(projection: PackageSearchProjection) projection.searchText ?? '', projection.exports.join('\n'), jobLines.join('\n'), + serviceLines.join('\n'), projection.appEntry ? `app ${projection.appEntry}` : projection.hasApp diff --git a/packages/worker/src/package-registry/service.node.test.ts b/packages/worker/src/package-registry/service.node.test.ts index 704fb73abc..8aedd9d064 100644 --- a/packages/worker/src/package-registry/service.node.test.ts +++ b/packages/worker/src/package-registry/service.node.test.ts @@ -1,5 +1,19 @@ import { beforeEach, expect, test, vi } from 'vitest' +function mockPackageServiceNamespace(): DurableObjectNamespace { + return { + idFromName(name: string) { + return { toString: () => name } as DurableObjectId + }, + get(id: DurableObjectId) { + return { + fetch: vi.fn(async () => new Response(JSON.stringify({ ok: true }))), + id, + } as unknown as DurableObjectStub + }, + } as DurableObjectNamespace +} + const mockModule = vi.hoisted(() => ({ buildPackageSearchProjection: vi.fn(), buildSavedPackageEmbedText: vi.fn(), @@ -9,8 +23,10 @@ const mockModule = vi.hoisted(() => ({ deleteSavedPackageVector: vi.fn(), getSavedPackageById: vi.fn(), insertSavedPackage: vi.fn(), + listSavedPackageServices: vi.fn(), listJobRowsByUserId: vi.fn(), loadPackageSourceBySourceId: vi.fn(), + packageServiceRpc: vi.fn(), syncJobManagerAlarm: vi.fn(), syncPackageJobsForPackage: vi.fn(), updateSavedPackage: vi.fn(), @@ -37,6 +53,13 @@ vi.mock('#worker/package-runtime/module-graph.ts', () => ({ buildKodyModuleBundle: vi.fn(), })) +vi.mock('#worker/package-runtime/package-service.ts', () => ({ + listSavedPackageServices: (...args: Array) => + mockModule.listSavedPackageServices(...args), + packageServiceRpc: (...args: Array) => + mockModule.packageServiceRpc(...args), +})) + vi.mock('./repo.ts', () => ({ deleteSavedPackage: (...args: Array) => mockModule.deleteSavedPackage(...args), @@ -82,6 +105,7 @@ const { deleteSavedPackageProjection, refreshSavedPackageProjection } = function createEnv() { return { APP_DB: {}, + PACKAGE_SERVICE_INSTANCE: mockPackageServiceNamespace(), } as Env } @@ -111,6 +135,17 @@ beforeEach(() => { mockModule.deleteSavedPackage.mockResolvedValue(undefined) mockModule.deleteSavedPackageVector.mockResolvedValue(undefined) mockModule.deleteJobRow.mockResolvedValue(undefined) + mockModule.listSavedPackageServices.mockResolvedValue({ + savedPackage: { + id: 'package-1', + kodyId: 'shade-automation', + }, + services: [], + }) + mockModule.packageServiceRpc.mockReturnValue({ + start: vi.fn().mockResolvedValue({ ok: true }), + stop: vi.fn().mockResolvedValue({ ok: true }), + }) }) test('refreshSavedPackageProjection resyncs the job manager after syncing package jobs', async () => { @@ -122,6 +157,12 @@ test('refreshSavedPackageProjection resyncs the job manager after syncing packag description: 'Shade automation package', tags: ['home', 'shades'], searchText: 'shade automation', + services: { + 'realtime-supervisor': { + entry: './src/services/realtime-supervisor.ts', + autoStart: true, + }, + }, jobs: { 'event-runner': { entry: './src/jobs/event-runner.ts', @@ -197,6 +238,7 @@ test('refreshSavedPackageProjection resyncs the job manager after syncing packag env, userId: 'user-1', }) + expect(mockModule.getSavedPackageById).toHaveBeenCalledTimes(1) expect( mockModule.syncJobManagerAlarm.mock.invocationCallOrder[0], ).toBeGreaterThan( @@ -208,8 +250,23 @@ test('deleteSavedPackageProjection resyncs the job manager after removing packag const env = createEnv() mockModule.getSavedPackageById.mockResolvedValue({ id: 'package-1', + kodyId: 'shade-automation', sourceId: 'source-1', }) + mockModule.listSavedPackageServices.mockResolvedValue({ + savedPackage: { + id: 'package-1', + kodyId: 'shade-automation', + }, + services: [ + { + name: 'realtime-supervisor', + entry: './src/services/realtime-supervisor.ts', + autoStart: true, + timeoutMs: null, + }, + ], + }) mockModule.listJobRowsByUserId.mockResolvedValue([ { id: 'job-1', source_id: 'source-1' }, { id: 'job-2', source_id: 'source-other' }, @@ -221,6 +278,15 @@ test('deleteSavedPackageProjection resyncs the job manager after removing packag packageId: 'package-1', }) + expect(mockModule.packageServiceRpc).toHaveBeenCalledWith({ + env, + userId: 'user-1', + packageId: 'package-1', + kodyId: 'shade-automation', + sourceId: 'source-1', + baseUrl: 'https://package-service.invalid', + serviceName: 'realtime-supervisor', + }) expect(mockModule.deleteJobRow).toHaveBeenCalledTimes(1) expect(mockModule.deleteJobRow).toHaveBeenCalledWith({}, 'user-1', 'job-1') expect(mockModule.deleteSavedPackage).toHaveBeenCalledWith( @@ -242,3 +308,114 @@ test('deleteSavedPackageProjection resyncs the job manager after removing packag mockModule.syncJobManagerAlarm.mock.invocationCallOrder[0], ).toBeGreaterThan(mockModule.deleteSavedPackage.mock.invocationCallOrder[0]) }) + +test('refreshSavedPackageProjection still syncs job manager when auto-start service startup fails', async () => { + const env = { + APP_DB: {}, + PACKAGE_SERVICE_INSTANCE: { + idFromName(name: string) { + return name as unknown as DurableObjectId + }, + get() { + return {} as DurableObjectStub + }, + }, + } as Env + const manifest = { + name: '@kentcdodds/shade-automation', + kody: { + id: 'shade-automation', + description: 'Shade automation package', + tags: ['home', 'shades'], + searchText: 'shade automation', + services: { + 'realtime-supervisor': { + entry: './src/services/realtime-supervisor.ts', + autoStart: true, + }, + }, + }, + } + mockModule.loadPackageSourceBySourceId.mockResolvedValue({ + manifest, + files: { 'package.json': '{}' }, + }) + mockModule.getSavedPackageById.mockResolvedValue({ + id: 'package-1', + userId: 'user-1', + name: '@kentcdodds/shade-automation', + kodyId: 'shade-automation', + description: 'Old description', + tags: ['home'], + searchText: null, + sourceId: 'source-1', + hasApp: false, + createdAt: '2026-04-20T00:00:00.000Z', + updatedAt: '2026-04-20T00:00:00.000Z', + }) + mockModule.packageServiceRpc.mockReturnValue({ + start: vi.fn().mockRejectedValue(new Error('service start failed')), + }) + + await refreshSavedPackageProjection({ + env, + baseUrl: 'https://heykody.dev', + userId: 'user-1', + packageId: 'package-1', + sourceId: 'source-1', + }) + + expect(mockModule.syncJobManagerAlarm).toHaveBeenCalledWith({ + env, + userId: 'user-1', + }) +}) + +test('deleteSavedPackageProjection still completes cleanup when service stop throws synchronously', async () => { + const env = createEnv() + mockModule.getSavedPackageById.mockResolvedValue({ + id: 'package-1', + kodyId: 'shade-automation', + sourceId: 'source-1', + }) + mockModule.listSavedPackageServices.mockResolvedValue({ + savedPackage: { + id: 'package-1', + kodyId: 'shade-automation', + }, + services: [ + { + name: 'realtime-supervisor', + entry: 'services/realtime-supervisor.ts', + autoStart: true, + timeoutMs: null, + }, + ], + }) + mockModule.listJobRowsByUserId.mockResolvedValue([ + { id: 'job-1', source_id: 'source-1' }, + ]) + mockModule.packageServiceRpc.mockImplementation(() => { + throw new Error('stub unavailable') + }) + + await deleteSavedPackageProjection({ + env, + userId: 'user-1', + packageId: 'package-1', + }) + + expect(mockModule.deleteJobRow).toHaveBeenCalledWith({}, 'user-1', 'job-1') + expect(mockModule.deleteSavedPackage).toHaveBeenCalledWith({}, { + userId: 'user-1', + packageId: 'package-1', + }) + expect(mockModule.deleteSavedPackageVector).toHaveBeenCalledWith( + env, + 'package-1', + ) + expect(mockModule.syncJobManagerAlarm).toHaveBeenCalledWith({ + env, + userId: 'user-1', + }) +}) diff --git a/packages/worker/src/package-registry/service.ts b/packages/worker/src/package-registry/service.ts index 4529516958..1ee61d3932 100644 --- a/packages/worker/src/package-registry/service.ts +++ b/packages/worker/src/package-registry/service.ts @@ -21,6 +21,10 @@ import { syncJobManagerAlarm } from '#worker/jobs/manager-client.ts' import { rebuildPublishedPackageArtifacts, } from '#worker/package-runtime/published-bundle-artifacts.ts' +import { + listSavedPackageServices, + packageServiceRpc, +} from '#worker/package-runtime/package-service.ts' function serializeTags(tags: Array) { return JSON.stringify(tags) @@ -144,6 +148,25 @@ export async function refreshSavedPackageProjection(input: { sourceId: input.sourceId, manifest: loaded.manifest, }) + for (const service of loaded.manifest.kody.services + ? Object.keys(loaded.manifest.kody.services) + : []) { + const definition = loaded.manifest.kody.services?.[service] + if (!definition?.autoStart) continue + try { + await packageServiceRpc({ + env: input.env, + userId: input.userId, + packageId: input.packageId, + kodyId: row.kody_id, + sourceId: row.source_id, + baseUrl: input.baseUrl, + serviceName: service, + }).start() + } catch { + // Auto-start failures should not block package job sync/alarm refresh. + } + } await syncJobManagerAlarm({ env: input.env, userId: input.userId, @@ -179,6 +202,30 @@ export async function deleteSavedPackageProjection(input: { packageId: input.packageId, }) if (savedPackage) { + const listedServices = await listSavedPackageServices({ + env: input.env, + userId: input.userId, + baseUrl: 'https://package-service.invalid', + packageId: input.packageId, + savedPackage, + }).catch(() => null) + if (listedServices) { + for (const service of listedServices.services) { + try { + await packageServiceRpc({ + env: input.env, + userId: input.userId, + packageId: savedPackage.id, + kodyId: savedPackage.kodyId, + sourceId: savedPackage.sourceId, + baseUrl: 'https://package-service.invalid', + serviceName: service.name, + }).stop() + } catch { + // Best-effort shutdown of orphaned services during package deletion. + } + } + } const existingRows = await listJobRowsByUserId( input.env.APP_DB, input.userId, diff --git a/packages/worker/src/package-registry/types.ts b/packages/worker/src/package-registry/types.ts index f24a15885c..afacc260a5 100644 --- a/packages/worker/src/package-registry/types.ts +++ b/packages/worker/src/package-registry/types.ts @@ -35,6 +35,16 @@ export const packageAppDefinitionSchema = z.object({ export type PackageAppDefinition = z.infer +export const packageServiceDefinitionSchema = z.object({ + entry: z.string().min(1), + autoStart: z.boolean().optional(), + timeoutMs: z.number().int().positive().max(300_000).optional(), +}) + +export type PackageServiceDefinition = z.infer< + typeof packageServiceDefinitionSchema +> + const packageExportConditionSchema = z .object({ import: z.string().min(1).optional(), @@ -65,6 +75,9 @@ export const authoredPackageKodySchema = z.object({ tags: z.array(z.string().min(1)).optional(), searchText: z.string().min(1).optional(), app: packageAppDefinitionSchema.optional(), + services: z + .record(z.string().min(1), packageServiceDefinitionSchema) + .optional(), jobs: z.record(z.string().min(1), packageJobDefinitionSchema).optional(), }) diff --git a/packages/worker/src/package-runtime/module-graph.node.test.ts b/packages/worker/src/package-runtime/module-graph.node.test.ts index 40a8e95e5c..798c157c40 100644 --- a/packages/worker/src/package-runtime/module-graph.node.test.ts +++ b/packages/worker/src/package-runtime/module-graph.node.test.ts @@ -682,3 +682,49 @@ test('buildKodyAppBundle rewrites dynamic kody runtime imports inside TypeScript "import('kody:runtime')", ) }) + +test('buildKodyAppBundle runtime module exports service helper', async () => { + mockModule.createWorker.mockResolvedValue(createBundleResult('runtime-service-helper')) + + await buildKodyAppBundle({ + env: { + APP_DB: {}, + REPO_SESSION: {}, + } as Env, + baseUrl: 'https://heykody.dev', + userId: 'user-1', + sourceFiles: { + 'package.json': JSON.stringify({ + name: '@kentcdodds/example-package', + exports: { + '.': './index.ts', + }, + kody: { + id: 'example-package', + description: 'Example package', + app: { + entry: 'app.ts', + }, + }, + }), + 'app.ts': `export default { + async fetch() { + return Response.json({ ok: true }) + }, +} +`, + 'index.ts': 'export const value = "ok"', + }, + entryPoint: 'app.ts', + cacheKey: null, + }) + + const firstCall = mockModule.createWorker.mock.calls.at(-1)?.[0] as + | { + files?: Record + } + | undefined + expect(firstCall?.files?.['.__kody_virtual__/runtime.js']).toContain( + 'export const service = runtime.service ?? null;', + ) +}) diff --git a/packages/worker/src/package-runtime/module-graph.ts b/packages/worker/src/package-runtime/module-graph.ts index 8de0798b1b..d318908d92 100644 --- a/packages/worker/src/package-runtime/module-graph.ts +++ b/packages/worker/src/package-runtime/module-graph.ts @@ -104,6 +104,8 @@ export const refreshAccessToken = runtime.refreshAccessToken; export const createAuthenticatedFetch = runtime.createAuthenticatedFetch; export const agentChatTurnStream = runtime.agentChatTurnStream; export const packageContext = runtime.packageContext ?? null; +export const serviceContext = runtime.serviceContext ?? null; +export const service = runtime.service ?? null; export default runtime; `.trim() @@ -518,6 +520,9 @@ export function createPublishedBundleArtifact(input: { packageId: string kodyId: string } | null + serviceContext?: { + serviceName: string + } | null }): PublishedBundleArtifact { return { version: 1, @@ -530,6 +535,7 @@ export function createPublishedBundleArtifact(input: { modules: input.modules, dependencies: input.dependencies, packageContext: input.packageContext ?? null, + serviceContext: input.serviceContext ?? null, createdAt: new Date().toISOString(), } } diff --git a/packages/worker/src/package-runtime/package-app.ts b/packages/worker/src/package-runtime/package-app.ts index b43f5d1c28..b2b8d6a2ff 100644 --- a/packages/worker/src/package-runtime/package-app.ts +++ b/packages/worker/src/package-runtime/package-app.ts @@ -21,6 +21,11 @@ import { } from './published-runtime-artifacts.ts' import { storageRunnerRpc } from '#worker/storage-runner.ts' import { packageRealtimeSessionRpc } from './realtime-session.ts' +import { + listSavedPackageServices, + normalizePackageServiceStatus, + packageServiceRpc, +} from './package-service.ts' const packageAppEntrypointName = 'PackageAppWorker' const packageAppRuntimeBindingName = 'KODY_RUNTIME' @@ -183,6 +188,24 @@ function createRealtimeProxy(runtimeBridge) { }; } +function createServicesProxy(runtimeBridge) { + return { + list: async () => await runtimeBridge.serviceList(), + get: async (serviceName) => + await runtimeBridge.serviceGet({ + serviceName, + }), + start: async (serviceName) => + await runtimeBridge.serviceStart({ + serviceName, + }), + stop: async (serviceName) => + await runtimeBridge.serviceStop({ + serviceName, + }), + }; +} + function createAuthenticatedFetchHelper(runtimeBridge) { return async function createAuthenticatedFetch(providerName) { return async (input, init) => @@ -316,6 +339,7 @@ function createRuntime(runtimeBridge, params, packageContext) { createAuthenticatedFetch: createAuthenticatedFetchHelper(runtimeBridge), agentChatTurnStream: createAgentChatTurnStream(runtimeBridge), realtime: createRealtimeProxy(runtimeBridge), + services: createServicesProxy(runtimeBridge), packageContext, }; } @@ -472,6 +496,18 @@ export class PackageAppRuntimeBridge extends WorkerEntrypoint< }) } + private getPackageServiceRpc(serviceName: string) { + return packageServiceRpc({ + env: this.env, + userId: this.ctx.props.userId, + packageId: this.ctx.props.packageId, + kodyId: this.ctx.props.kodyId, + sourceId: this.ctx.props.sourceId, + baseUrl: this.ctx.props.baseUrl, + serviceName, + }) + } + async callCapability(input: { name: string; args?: unknown }) { const name = input.name.trim() switch (name) { @@ -668,6 +704,51 @@ export class PackageAppRuntimeBridge extends WorkerEntrypoint< reason: input.reason ?? undefined, }) } + + async serviceList() { + const result = await listSavedPackageServices({ + env: this.env, + userId: this.ctx.props.userId, + baseUrl: this.ctx.props.baseUrl, + packageId: this.ctx.props.packageId, + }) + const services = await Promise.all( + result.services.map(async (service) => { + let status = 'unknown' + try { + status = normalizePackageServiceStatus( + await this.getPackageServiceRpc(service.name).status(), + ).status + } catch { + // Keep the rest of the service list usable if one status lookup fails. + } + return { + name: service.name, + entry: service.entry, + auto_start: service.autoStart, + timeout_ms: service.timeoutMs ?? null, + status, + } + }), + ) + return { + package_id: result.savedPackage.id, + kody_id: result.savedPackage.kodyId, + services, + } + } + + async serviceGet(input: { serviceName: string }) { + return await this.getPackageServiceRpc(input.serviceName).status() + } + + async serviceStart(input: { serviceName: string }) { + return await this.getPackageServiceRpc(input.serviceName).start() + } + + async serviceStop(input: { serviceName: string }) { + return await this.getPackageServiceRpc(input.serviceName).stop() + } } export async function buildPackageAppWorker(input: { diff --git a/packages/worker/src/package-runtime/package-service.node.test.ts b/packages/worker/src/package-runtime/package-service.node.test.ts new file mode 100644 index 0000000000..29be519fd6 --- /dev/null +++ b/packages/worker/src/package-runtime/package-service.node.test.ts @@ -0,0 +1,123 @@ +import { expect, test } from 'vitest' +import { buildPackageServiceStorageId } from './package-service.ts' + +test('buildPackageServiceStorageId creates stable per-service storage ids', () => { + expect(buildPackageServiceStorageId('package-1', 'realtime-supervisor')).toBe( + 'service:package-1:realtime-supervisor', + ) + expect(buildPackageServiceStorageId('package-1', 'guild sync')).toBe( + 'service:package-1:guild%20sync', + ) + expect(buildPackageServiceStorageId('package-1', 'a:b/c')).toBe( + 'service:package-1:a%3Ab%2Fc', + ) +}) + +test('package service runtime source clears stale in-flight state on restore', async () => { + const source = await import('./package-service.ts') + const fileText = await import('node:fs/promises').then((fs) => + fs.readFile(new URL('./package-service.ts', import.meta.url), 'utf8'), + ) + expect(fileText).toContain("this.stateSnapshot.status === 'running'") + expect(fileText).toContain("this.stateSnapshot.status === 'stopping'") + expect(fileText).toContain("this.stateSnapshot.status = 'stopped'") + expect(fileText).toContain('this.stateSnapshot.currentRunId = null') + expect(fileText).toContain('this.stateSnapshot.stopRequested = false') + expect(source.buildPackageServiceStorageId).toBeTypeOf('function') +}) + +test('package service runtime preserves explicit stop requests after a run ends', async () => { + const fileText = await import('node:fs/promises').then((fs) => + fs.readFile(new URL('./package-service.ts', import.meta.url), 'utf8'), + ) + expect(fileText).toContain( + 'const stopRequested = this.stateSnapshot.stopRequested', + ) + expect(fileText).toContain( + 'this.stateSnapshot.stopRequested = false', + ) + expect(fileText).toContain( + 'await this.finalizeServiceRun({', + ) + expect(fileText).toContain("if (stopRequested) {\n\t\t\tawait this.clearAlarm()") + expect(fileText).toContain('!this.stateSnapshot.stopRequested') +}) + +test('package service runtime restart clears a pending stop request', async () => { + const fileText = await import('node:fs/promises').then((fs) => + fs.readFile(new URL('./package-service.ts', import.meta.url), 'utf8'), + ) + expect(fileText).toContain('if (this.stateSnapshot.currentRunId) {') + expect(fileText).toContain('this.stateSnapshot.stopRequested = false') + expect(fileText).toContain('already_running: true') +}) + +test('package service runtime re-arms auto-start after unplanned exit', async () => { + const fileText = await import('node:fs/promises').then((fs) => + fs.readFile(new URL('./package-service.ts', import.meta.url), 'utf8'), + ) + expect(fileText).toContain('loaded.serviceDefinition?.autoStart') + expect(fileText).toContain('!this.stateSnapshot.nextAlarmAt') + expect(fileText).toContain('const packageServiceRetryDelayMs = 5_000') + expect(fileText).toContain('buildPackageServiceRetryTime()') + expect(fileText).toContain('runAt: buildPackageServiceRetryTime(),') + expect(fileText).toContain("source: 'auto-start'") + expect(fileText).toContain( + "this.stateSnapshot.status = 'error'", + ) + expect(fileText).toContain( + 'this.stateSnapshot.autoStart &&\n\t\t\t\t!this.stateSnapshot.stopRequested &&\n\t\t\t\t!this.stateSnapshot.nextAlarmAt', + ) +}) + +test('package service runtime refreshes manifest-backed service settings for alarms and status', async () => { + const fileText = await import('node:fs/promises').then((fs) => + fs.readFile(new URL('./package-service.ts', import.meta.url), 'utf8'), + ) + expect(fileText).toContain('const loaded = await loadSavedPackageService({') + expect(fileText).toContain('this.stateSnapshot.binding = loaded.resolvedBinding') + expect(fileText).toContain( + 'this.stateSnapshot.timeoutMs = loaded.serviceDefinition?.timeoutMs ?? null', + ) + expect(fileText).toContain("'timeoutMs' in overrides") + expect(fileText).toContain( + 'const binding = loaded?.resolvedBinding ?? this.stateSnapshot.binding ?? input.binding', + ) + expect(fileText).toContain( + 'loaded = await this.initializeBinding(input.binding, {', + ) + expect(fileText).toContain('this.buildServiceStatusResponse(') + expect(fileText).toContain('binding,') + expect(fileText).toContain('loaded?.serviceDefinition') + expect(fileText).not.toContain( + 'if (!this.stateSnapshot.stopRequested) {\n\t\t\t\tawait this.handleStartRequest({ binding: loaded.resolvedBinding })', + ) +}) + +test('package service runtime schedules auto-start on save path instead of read-only status calls', async () => { + const fileText = await import('node:fs/promises').then((fs) => + fs.readFile(new URL('./package-service.ts', import.meta.url), 'utf8'), + ) + expect(fileText).toContain('options?: { armAutoStart?: boolean }') + expect(fileText).toContain('options?.armAutoStart') + expect(fileText).toContain('armAutoStart: true') + expect(fileText).toContain("source: 'auto-start'") + expect(fileText).toContain( + 'const binding = loaded?.resolvedBinding ?? this.stateSnapshot.binding ?? input.binding', + ) + expect(fileText).toContain( + 'options?.armAutoStart &&', + ) +}) + +test('package service runtime persists restored state and surfaces RPC errors', async () => { + const fileText = await import('node:fs/promises').then((fs) => + fs.readFile(new URL('./package-service.ts', import.meta.url), 'utf8'), + ) + expect(fileText).toContain('await this.persistState()') + expect(fileText).toContain('async function readPackageServiceRpcResponse(') + expect(fileText).toContain('if (!response.ok) {') + expect(fileText).toContain( + 'text || `Package service request failed with status ${response.status}.`', + ) +}) diff --git a/packages/worker/src/package-runtime/package-service.ts b/packages/worker/src/package-runtime/package-service.ts new file mode 100644 index 0000000000..82a87a41a5 --- /dev/null +++ b/packages/worker/src/package-runtime/package-service.ts @@ -0,0 +1,778 @@ +import * as Sentry from '@sentry/cloudflare' +import { DurableObject } from 'cloudflare:workers' +import { z } from 'zod' +import { createMcpCallerContext } from '#mcp/context.ts' +import { + getPackageServiceEntryPath, + listPackageServices, +} from '#worker/package-registry/manifest.ts' +import { getSavedPackageById } from '#worker/package-registry/repo.ts' +import { loadPackageSourceBySourceId } from '#worker/package-registry/source.ts' +import { buildSentryOptions } from '#worker/sentry-options.ts' + +const serviceStateStorageKey = 'package-service-state' +const packageServiceRetryDelayMs = 5_000 + +export type PackageServiceBindingState = { + userId: string + packageId: string + kodyId: string + sourceId: string + baseUrl: string + serviceName: string +} + +type PackageServiceState = { + binding: PackageServiceBindingState | null + autoStart: boolean + timeoutMs: number | null + stopRequested: boolean + currentRunId: string | null + nextAlarmAt: string | null + nextAlarmSource: 'service' | 'auto-start' | null + lastStartedAt: string | null + lastStoppedAt: string | null + status: 'idle' | 'running' | 'stopping' | 'stopped' | 'error' + lastError: string | null + lastResult: unknown + lastRunFinishedAt: string | null +} + +type PackageServiceRunResult = { + ok: boolean + run_id: string + started_at: string + status: 'running' + already_running?: boolean +} + +export const packageServiceStatusSchema = z.object({ + package_id: z.string(), + kody_id: z.string(), + service_name: z.string(), + status: z.enum(['idle', 'running', 'stopping', 'stopped', 'error']), + auto_start: z.boolean(), + timeout_ms: z.number().int().positive().nullable(), + stop_requested: z.boolean(), + active_run_id: z.string().nullable(), + next_alarm_at: z.string().nullable(), + last_error: z.string().nullable(), + last_started_at: z.string().nullable(), + last_stopped_at: z.string().nullable(), + last_run_finished_at: z.string().nullable(), + last_result: z.unknown(), +}) + +export type PackageServiceStatusRecord = z.infer< + typeof packageServiceStatusSchema +> + +export function normalizePackageServiceStatus( + input: unknown, +): PackageServiceStatusRecord { + const result = packageServiceStatusSchema.safeParse(input) + if (!result.success) { + throw new Error(z.prettifyError(result.error)) + } + return result.data +} + +function createInitialPackageServiceState(): PackageServiceState { + return { + binding: null, + autoStart: false, + timeoutMs: null, + stopRequested: false, + currentRunId: null, + nextAlarmAt: null, + nextAlarmSource: null, + lastStartedAt: null, + lastStoppedAt: null, + status: 'idle', + lastError: null, + lastResult: null, + lastRunFinishedAt: null, + } +} + +function buildPackageServiceName(input: { + userId: string + packageId: string + serviceName: string +}) { + return JSON.stringify([input.userId, input.packageId, input.serviceName]) +} + +function getPackageServiceNamespace(env: Env) { + return env.PACKAGE_SERVICE_INSTANCE +} + +function getPackageServiceStub(input: { + env: Env + userId: string + packageId: string + serviceName: string +}) { + const namespace = getPackageServiceNamespace(input.env) + if (!namespace) { + throw new Error('Missing PACKAGE_SERVICE_INSTANCE binding.') + } + const id = namespace.idFromName( + buildPackageServiceName({ + userId: input.userId, + packageId: input.packageId, + serviceName: input.serviceName, + }), + ) + return namespace.get(id) +} + +export function buildPackageServiceStorageId( + packageId: string, + serviceName: string, +) { + return `service:${encodeURIComponent(packageId)}:${encodeURIComponent(serviceName)}` +} + +async function loadSavedPackageService(input: { + env: Env + binding: PackageServiceBindingState +}) { + const savedPackage = await getSavedPackageById(input.env.APP_DB, { + userId: input.binding.userId, + packageId: input.binding.packageId, + }) + if (!savedPackage) { + throw new Error('Saved package was not found for package service operations.') + } + const packageSource = await loadPackageSourceBySourceId({ + env: input.env, + baseUrl: input.binding.baseUrl, + userId: input.binding.userId, + sourceId: savedPackage.sourceId, + }) + const resolvedBinding: PackageServiceBindingState = { + ...input.binding, + kodyId: savedPackage.kodyId, + sourceId: savedPackage.sourceId, + } + const serviceDefinition = + packageSource.manifest.kody.services?.[input.binding.serviceName] + const serviceEntry = getPackageServiceEntryPath({ + manifest: packageSource.manifest, + serviceName: input.binding.serviceName, + }) + if (!serviceEntry) { + throw new Error( + `Saved package "${packageSource.manifest.kody.id}" does not define service "${input.binding.serviceName}".`, + ) + } + return { + resolvedBinding, + savedPackage, + packageSource, + serviceDefinition, + serviceEntry, + } +} + +async function readPackageServiceRpcResponse( + response: Response, +): Promise { + const text = await response.text() + if (!response.ok) { + throw new Error( + text || `Package service request failed with status ${response.status}.`, + ) + } + try { + return JSON.parse(text) as T + } catch { + throw new Error('Package service returned an invalid JSON response.') + } +} + +function buildPackageServiceRetryTime() { + return new Date(Date.now() + packageServiceRetryDelayMs) +} + +class PackageServiceInstanceBase extends DurableObject { + private stateSnapshot: PackageServiceState = createInitialPackageServiceState() + private activeRunPromise: Promise | null = null + + constructor(state: DurableObjectState, env: Env) { + super(state, env) + this.ctx.blockConcurrencyWhile(async () => { + await this.restoreState() + }) + } + + private async restoreState() { + const stored = + await this.ctx.storage.get(serviceStateStorageKey) + if (!stored) return + this.stateSnapshot = stored + if ( + this.stateSnapshot.currentRunId && + (this.stateSnapshot.status === 'running' || + this.stateSnapshot.status === 'stopping') + ) { + // Background execution does not survive Durable Object eviction, so a + // restored in-flight run must be downgraded to a recoverable stopped + // state. + const wasExplicitlyStopping = this.stateSnapshot.status === 'stopping' + this.stateSnapshot.currentRunId = null + this.stateSnapshot.stopRequested = wasExplicitlyStopping + this.stateSnapshot.status = 'stopped' + this.stateSnapshot.lastStoppedAt = new Date().toISOString() + await this.persistState() + if ( + this.stateSnapshot.autoStart && + !this.stateSnapshot.stopRequested && + this.stateSnapshot.binding + ) { + await this.scheduleAlarm({ + runAt: buildPackageServiceRetryTime(), + source: 'auto-start', + }) + } + } + } + + private async persistState() { + await this.ctx.storage.put(serviceStateStorageKey, this.stateSnapshot) + } + + private async initializeBinding( + binding: PackageServiceBindingState, + options?: { armAutoStart?: boolean }, + ) { + const existing = this.stateSnapshot.binding + if ( + existing && + (existing.userId !== binding.userId || + existing.packageId !== binding.packageId || + existing.serviceName !== binding.serviceName) + ) { + throw new Error('Package service instance binding mismatch.') + } + const loaded = await loadSavedPackageService({ + env: this.env, + binding, + }) + this.stateSnapshot.binding = loaded.resolvedBinding + this.stateSnapshot.autoStart = loaded.serviceDefinition?.autoStart ?? false + this.stateSnapshot.timeoutMs = loaded.serviceDefinition?.timeoutMs ?? null + await this.persistState() + if ( + options?.armAutoStart && + !existing && + this.stateSnapshot.autoStart && + !this.stateSnapshot.nextAlarmAt + ) { + await this.scheduleAlarm({ + runAt: new Date(), + source: 'auto-start', + }) + } + return loaded + } + + private async scheduleAlarm(input: { + runAt: Date | string + source?: 'service' | 'auto-start' + }) { + const runAtDate = + typeof input.runAt === 'string' ? new Date(input.runAt) : input.runAt + if (Number.isNaN(runAtDate.getTime())) { + throw new Error('Invalid runAt value provided to setAlarm.') + } + const scheduledAt = runAtDate.toISOString() + await this.ctx.storage.setAlarm(runAtDate) + this.stateSnapshot.nextAlarmAt = scheduledAt + this.stateSnapshot.nextAlarmSource = input.source ?? 'service' + await this.persistState() + return { + ok: true, + scheduled_at: scheduledAt, + } + } + + private async clearAlarm() { + await this.ctx.storage.deleteAlarm().catch(() => { + // Best effort cleanup. + }) + this.stateSnapshot.nextAlarmAt = null + this.stateSnapshot.nextAlarmSource = null + await this.persistState() + return { + ok: true, + } + } + + private buildServiceStatusResponse( + binding: PackageServiceBindingState, + overrides?: { + autoStart?: boolean + timeoutMs?: number | null + }, + ) { + const autoStart = + overrides && 'autoStart' in overrides + ? overrides.autoStart + : this.stateSnapshot.autoStart + const timeoutMs = + overrides && 'timeoutMs' in overrides + ? overrides.timeoutMs + : this.stateSnapshot.timeoutMs + return { + package_id: binding.packageId, + kody_id: binding.kodyId, + service_name: binding.serviceName, + status: this.stateSnapshot.status, + auto_start: autoStart, + timeout_ms: timeoutMs, + stop_requested: this.stateSnapshot.stopRequested, + active_run_id: this.stateSnapshot.currentRunId, + next_alarm_at: this.stateSnapshot.nextAlarmAt, + last_error: this.stateSnapshot.lastError, + last_started_at: this.stateSnapshot.lastStartedAt, + last_stopped_at: this.stateSnapshot.lastStoppedAt, + last_run_finished_at: this.stateSnapshot.lastRunFinishedAt, + last_result: this.stateSnapshot.lastResult, + } + } + + private async finalizeServiceRun(input: { + runId: string + nextStatus: PackageServiceState['status'] + lastResult?: unknown + lastError?: string | null + }) { + if (this.stateSnapshot.currentRunId !== input.runId) return + const stopRequested = this.stateSnapshot.stopRequested + this.stateSnapshot.status = input.nextStatus + this.stateSnapshot.currentRunId = null + this.stateSnapshot.stopRequested = false + if ('lastResult' in input) { + this.stateSnapshot.lastResult = input.lastResult ?? null + } + if ('lastError' in input) { + this.stateSnapshot.lastError = input.lastError ?? null + } + this.stateSnapshot.lastRunFinishedAt = new Date().toISOString() + this.stateSnapshot.lastStoppedAt = this.stateSnapshot.lastRunFinishedAt + await this.persistState() + if (stopRequested) { + await this.clearAlarm() + } else if ( + this.stateSnapshot.autoStart && + !this.stateSnapshot.nextAlarmAt + ) { + await this.scheduleAlarm({ + runAt: buildPackageServiceRetryTime(), + source: 'auto-start', + }) + } + } + + private async runServiceInBackground(input: { + binding: PackageServiceBindingState + runId: string + loaded?: Awaited> + }) { + try { + const loaded = + input.loaded ?? + (await loadSavedPackageService({ + env: this.env, + binding: input.binding, + })) + const storageId = buildPackageServiceStorageId( + input.binding.packageId, + input.binding.serviceName, + ) + this.stateSnapshot.binding = loaded.resolvedBinding + this.stateSnapshot.autoStart = loaded.serviceDefinition?.autoStart ?? false + this.stateSnapshot.timeoutMs = loaded.serviceDefinition?.timeoutMs ?? null + await this.persistState() + const result = await this.runService(loaded.resolvedBinding, { + getStatus: async () => + this.buildServiceStatusResponse(loaded.resolvedBinding, { + autoStart: loaded.serviceDefinition?.autoStart ?? false, + timeoutMs: loaded.serviceDefinition?.timeoutMs ?? null, + }), + shouldStop: async () => this.stateSnapshot.stopRequested, + setAlarm: async (runAt) => + (await this.scheduleAlarm({ + runAt, + source: 'service', + })) as { ok: true; scheduled_at: string }, + clearAlarm: async () => + (await this.clearAlarm()) as { ok: true }, + packageContext: { + packageId: loaded.savedPackage.id, + kodyId: loaded.savedPackage.kodyId, + }, + loaded, + executorTimeoutMs: loaded.serviceDefinition?.timeoutMs ?? 300_000, + storageId, + }) + await this.finalizeServiceRun({ + runId: input.runId, + nextStatus: 'stopped', + lastResult: result, + }) + } catch (error) { + const errorMessage = + error instanceof Error ? error.message : String(error) + await this.finalizeServiceRun({ + runId: input.runId, + nextStatus: this.stateSnapshot.stopRequested ? 'stopped' : 'error', + lastError: errorMessage, + }) + } finally { + if (this.activeRunPromise) { + this.activeRunPromise = null + } + } + } + + private async runService( + binding: PackageServiceBindingState, + runtime: { + getStatus: () => Promise> + shouldStop: () => Promise + setAlarm: (runAt: Date | string) => Promise<{ ok: true; scheduled_at: string }> + clearAlarm: () => Promise<{ ok: true }> + packageContext: { + packageId: string + kodyId: string + } + loaded: Awaited> + executorTimeoutMs: number + storageId: string + }, + ) { + const [{ runBundledModuleWithRegistry }, { buildKodyModuleBundle }, { loadPublishedBundleArtifactByIdentity }] = + await Promise.all([ + import('#mcp/run-codemode-registry.ts'), + import('./module-graph.ts'), + import('./published-bundle-artifacts.ts'), + ]) + const artifact = await loadPublishedBundleArtifactByIdentity({ + env: this.env, + userId: binding.userId, + sourceId: binding.sourceId, + kind: 'service', + artifactName: binding.serviceName, + entryPoint: runtime.loaded.serviceEntry, + }) + const bundle = + artifact?.artifact ?? + (await buildKodyModuleBundle({ + env: this.env, + baseUrl: binding.baseUrl, + userId: binding.userId, + sourceFiles: runtime.loaded.packageSource.files, + entryPoint: runtime.loaded.serviceEntry, + })) + const callerContext = createMcpCallerContext({ + baseUrl: binding.baseUrl, + user: { + userId: binding.userId, + email: '', + displayName: `package:${binding.packageId}`, + }, + storageContext: { + sessionId: null, + appId: binding.packageId, + storageId: runtime.storageId, + }, + }) + const result = await runBundledModuleWithRegistry( + this.env, + callerContext, + { + mainModule: bundle.mainModule, + modules: bundle.modules, + }, + undefined, + { + packageContext: runtime.packageContext, + serviceContext: { + serviceName: binding.serviceName, + }, + serviceTools: { + getStatus: runtime.getStatus, + shouldStop: runtime.shouldStop, + setAlarm: runtime.setAlarm, + clearAlarm: runtime.clearAlarm, + }, + storageTools: { + userId: binding.userId, + storageId: runtime.storageId, + writable: true, + }, + executorTimeoutMs: runtime.executorTimeoutMs, + }, + ) + if (result.error) { + const rawError: unknown = result.error + const errorMessage = + typeof rawError === 'string' + ? rawError + : typeof rawError === 'object' && + rawError !== null && + 'message' in rawError && + typeof rawError.message === 'string' + ? rawError.message + : String(rawError) + throw new Error(errorMessage) + } + return result.result ?? null + } + + private async handleStartRequest(input: { + binding: PackageServiceBindingState + }) { + const loaded = await this.initializeBinding(input.binding, { + armAutoStart: true, + }) + if (this.stateSnapshot.currentRunId) { + this.stateSnapshot.stopRequested = false + this.stateSnapshot.status = 'running' + await this.persistState() + return Response.json({ + ok: true, + run_id: this.stateSnapshot.currentRunId, + started_at: this.stateSnapshot.lastStartedAt ?? new Date().toISOString(), + status: 'running', + already_running: true, + } satisfies PackageServiceRunResult) + } + const startedAt = new Date().toISOString() + const runId = crypto.randomUUID() + this.stateSnapshot.stopRequested = false + this.stateSnapshot.currentRunId = runId + this.stateSnapshot.status = 'running' + this.stateSnapshot.lastStartedAt = startedAt + this.stateSnapshot.lastError = null + await this.persistState() + const task = this.runServiceInBackground({ + binding: loaded.resolvedBinding, + runId, + loaded, + }) + this.activeRunPromise = task + this.ctx.waitUntil(task) + return Response.json({ + ok: true, + run_id: runId, + started_at: startedAt, + status: 'running', + } satisfies PackageServiceRunResult) + } + + private async handleStatusRequest(input: { + binding: PackageServiceBindingState + }) { + let loaded: Awaited> | undefined + try { + loaded = await loadSavedPackageService({ + env: this.env, + binding: input.binding, + }) + } catch { + loaded = undefined + } + const binding = loaded?.resolvedBinding ?? this.stateSnapshot.binding ?? input.binding + return Response.json( + this.buildServiceStatusResponse(binding, { + autoStart: loaded?.serviceDefinition?.autoStart ?? this.stateSnapshot.autoStart, + timeoutMs: loaded?.serviceDefinition?.timeoutMs ?? this.stateSnapshot.timeoutMs, + }), + ) + } + + private async handleStopRequest(input: { + binding: PackageServiceBindingState + }) { + try { + await this.initializeBinding(input.binding) + } catch { + // Allow an in-flight service to be stopped even if its package/source was removed. + this.stateSnapshot.binding ??= input.binding + } + if (this.stateSnapshot.currentRunId) { + this.stateSnapshot.stopRequested = true + this.stateSnapshot.status = 'stopping' + this.stateSnapshot.lastStoppedAt = new Date().toISOString() + } else { + this.stateSnapshot.stopRequested = false + this.stateSnapshot.status = 'stopped' + } + await this.clearAlarm() + return Response.json({ + ok: true, + }) + } + + async fetch(request: Request) { + const url = new URL(request.url) + const body = (await request.json().catch(() => null)) as + | { binding?: PackageServiceBindingState } + | null + const binding = body?.binding + if (!binding) { + return new Response('Missing package service binding.', { status: 400 }) + } + if (request.method === 'POST' && url.pathname.endsWith('/start')) { + return await this.handleStartRequest({ binding }) + } + if (request.method === 'POST' && url.pathname.endsWith('/status')) { + return await this.handleStatusRequest({ binding }) + } + if (request.method === 'POST' && url.pathname.endsWith('/stop')) { + return await this.handleStopRequest({ binding }) + } + return new Response('Not found', { status: 404 }) + } + + async alarm() { + const binding = this.stateSnapshot.binding + if (!binding) return + const alarmSource = this.stateSnapshot.nextAlarmSource ?? 'service' + this.stateSnapshot.nextAlarmAt = null + this.stateSnapshot.nextAlarmSource = null + await this.persistState() + if (this.stateSnapshot.currentRunId) return + try { + const loaded = await loadSavedPackageService({ + env: this.env, + binding, + }) + this.stateSnapshot.binding = loaded.resolvedBinding + this.stateSnapshot.autoStart = loaded.serviceDefinition?.autoStart ?? false + this.stateSnapshot.timeoutMs = + loaded.serviceDefinition?.timeoutMs ?? null + await this.persistState() + if ( + !this.stateSnapshot.stopRequested && + (alarmSource === 'service' || this.stateSnapshot.autoStart) + ) { + const startedAt = new Date().toISOString() + const runId = crypto.randomUUID() + this.stateSnapshot.currentRunId = runId + this.stateSnapshot.status = 'running' + this.stateSnapshot.lastStartedAt = startedAt + this.stateSnapshot.lastError = null + await this.persistState() + const task = this.runServiceInBackground({ + binding: loaded.resolvedBinding, + runId, + loaded, + }) + this.activeRunPromise = task + this.ctx.waitUntil(task) + } + } catch (error) { + this.stateSnapshot.lastError = + error instanceof Error ? error.message : String(error) + this.stateSnapshot.status = 'error' + await this.persistState() + if ( + this.stateSnapshot.autoStart && + !this.stateSnapshot.stopRequested && + !this.stateSnapshot.nextAlarmAt + ) { + await this.scheduleAlarm({ + runAt: buildPackageServiceRetryTime(), + source: alarmSource, + }) + } + } + } +} + +export const PackageServiceInstance = Sentry.instrumentDurableObjectWithSentry( + (env: Env) => buildSentryOptions(env), + PackageServiceInstanceBase, +) + +export function packageServiceRpc(input: { + env: Env + userId: string + packageId: string + kodyId: string + sourceId: string + baseUrl: string + serviceName: string +}) { + const binding: PackageServiceBindingState = { + userId: input.userId, + packageId: input.packageId, + kodyId: input.kodyId, + sourceId: input.sourceId, + baseUrl: input.baseUrl, + serviceName: input.serviceName, + } + const stub = getPackageServiceStub(input) + async function callService(path: string): Promise { + const response = await stub.fetch( + new Request(`https://package-service.invalid${path}`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ binding }), + }), + ) + return await readPackageServiceRpcResponse(response) + } + return { + async start() { + return await callService('/service/start') + }, + async status() { + return await callService< + ReturnType + >('/service/status') + }, + async stop() { + return await callService<{ ok: true }>('/service/stop') + }, + } +} + +export async function listSavedPackageServices(input: { + env: Env + userId: string + baseUrl: string + packageId: string + savedPackage?: { + id: string + sourceId: string + kodyId?: string + } +}) { + const savedPackage = + input.savedPackage ?? + (await getSavedPackageById(input.env.APP_DB, { + userId: input.userId, + packageId: input.packageId, + })) + if (!savedPackage) { + throw new Error('Saved package was not found.') + } + const loaded = await loadPackageSourceBySourceId({ + env: input.env, + baseUrl: input.baseUrl, + userId: input.userId, + sourceId: savedPackage.sourceId, + }) + return { + savedPackage, + services: listPackageServices(loaded.manifest), + } +} diff --git a/packages/worker/src/package-runtime/published-bundle-artifacts.ts b/packages/worker/src/package-runtime/published-bundle-artifacts.ts index 93238cc2ec..768f0b9a17 100644 --- a/packages/worker/src/package-runtime/published-bundle-artifacts.ts +++ b/packages/worker/src/package-runtime/published-bundle-artifacts.ts @@ -1,4 +1,7 @@ -import { getPackageAppEntryPath } from '#worker/package-registry/manifest.ts' +import { + getPackageAppEntryPath, + listPackageServices, +} from '#worker/package-registry/manifest.ts' import { type AuthoredPackageJson, type SavedPackageRecord, @@ -168,6 +171,7 @@ export async function persistPublishedBundleArtifact(input: PersistPublishedBund modules: input.modules, dependencies: input.dependencies, packageContext: input.packageContext ?? null, + serviceContext: null, createdAt: new Date().toISOString(), } const existing = await getPublishedBundleArtifactByIdentity(input.env.APP_DB, { @@ -292,6 +296,26 @@ export async function rebuildPublishedPackageArtifacts(input: { }) } } + for (const service of listPackageServices(input.manifest)) { + const bundle = await input.buildModuleBundle({ + entryPoint: service.entry, + }) + await persistPublishedBundleArtifact({ + env: input.env, + userId: input.userId, + source: input.source, + kind: 'service', + artifactName: service.name, + entryPoint: service.entry, + mainModule: bundle.mainModule, + modules: bundle.modules, + dependencies: bundle.dependencies ?? fallbackDependencies, + packageContext: { + packageId: input.savedPackage.id, + kodyId: input.savedPackage.kodyId, + }, + }) + } for (const [exportName, exportTarget] of Object.entries( input.manifest.exports, ) as Array<[string, AuthoredPackageJson['exports'][string]]>) { diff --git a/packages/worker/src/package-runtime/published-runtime-artifacts.ts b/packages/worker/src/package-runtime/published-runtime-artifacts.ts index 29c9bc8802..90ad5cf4b9 100644 --- a/packages/worker/src/package-runtime/published-runtime-artifacts.ts +++ b/packages/worker/src/package-runtime/published-runtime-artifacts.ts @@ -8,7 +8,7 @@ const sourceSnapshotPrefix = 'source-snapshot' const sourceManifestSnapshotPrefix = 'source-manifest-snapshot' const bundleArtifactPrefix = 'bundle-artifact' -export type BundleArtifactKind = 'app' | 'job' | 'module' +export type BundleArtifactKind = 'app' | 'job' | 'module' | 'service' export type BundleArtifactDependency = { sourceId: string @@ -63,9 +63,20 @@ export type PublishedBundleArtifact = { packageId: string kodyId: string } | null + serviceContext: { + serviceName: string + } | null createdAt: string } +type StoredPublishedBundleArtifact = Omit< + PublishedBundleArtifact, + 'modules' | 'serviceContext' +> & { + modules: Record + serviceContext?: PublishedBundleArtifact['serviceContext'] +} + function getBundleArtifactsKv(env: Env) { const kv = (env as Env & { BUNDLE_ARTIFACTS_KV?: KVNamespace }).BUNDLE_ARTIFACTS_KV if (!kv) { @@ -403,9 +414,7 @@ export async function readPublishedBundleArtifact(input: { }) { const stored = await getBundleArtifactsKv(input.env).get(input.kvKey, 'json') if (!stored || typeof stored !== 'object') return null - const artifact = stored as Omit & { - modules: Record - } + const artifact = stored as StoredPublishedBundleArtifact if ( artifact.version !== bundleArtifactVersion || typeof artifact.modules !== 'object' || @@ -415,6 +424,7 @@ export async function readPublishedBundleArtifact(input: { } return { ...artifact, + serviceContext: artifact.serviceContext ?? null, modules: deserializeWorkerLoaderModules(artifact.modules), } satisfies PublishedBundleArtifact } diff --git a/packages/worker/src/repo/checks.ts b/packages/worker/src/repo/checks.ts index b7fe61a10a..e674e6bc1d 100644 --- a/packages/worker/src/repo/checks.ts +++ b/packages/worker/src/repo/checks.ts @@ -1,5 +1,6 @@ import { getPackageAppEntryPath, + listPackageServices, normalizePackageWorkspacePath, parseAuthoredPackageJson, resolvePackageExportPath, @@ -197,6 +198,16 @@ declare function createAuthenticatedFetch( providerName: string, ): Promise<(input: string | URL | Request, init?: RequestInit) => Promise>; declare function agentChatTurnStream(input: KodyCapabilityArgs): AsyncIterable; +declare const packageContext: { packageId: string; kodyId: string } | null; +declare const serviceContext: { serviceName: string } | null; +declare const service: + | { + getStatus(): Promise; + shouldStop(): Promise; + setAlarm(runAt: string | Date): Promise; + clearAlarm(): Promise; + } + | null; ${ input?.includeStorage === true ? ` @@ -252,6 +263,9 @@ function collectPackageTypecheckTargets( for (const job of Object.values(manifest.kody.jobs ?? {})) { remember(job.entry, true) } + for (const service of listPackageServices(manifest)) { + remember(service.entry, true) + } return Array.from(targets.values()) } @@ -390,7 +404,7 @@ function formatBundleCheckMessage(input: { .join(', ')}.` } if (input.targetCount === 0) { - return 'Package defines no app entry, exports, or jobs to bundle.' + return 'Package defines no app entry, exports, jobs, or services to bundle.' } return `Resolved ${input.targetCount} package runtime entrypoint(s) for bundling.` } diff --git a/packages/worker/worker-configuration.d.ts b/packages/worker/worker-configuration.d.ts index 0bc804b0ab..9a78db5e41 100644 --- a/packages/worker/worker-configuration.d.ts +++ b/packages/worker/worker-configuration.d.ts @@ -4,7 +4,7 @@ declare namespace Cloudflare { interface GlobalProps { mainModule: typeof import("./src/index"); - durableNamespaces: "MCP" | "ChatAgent" | "HomeConnectorSession" | "HomeMCP" | "JobManager" | "StorageRunner" | "AgentTurnRunner" | "RepoSession"; + durableNamespaces: "MCP" | "ChatAgent" | "HomeConnectorSession" | "HomeMCP" | "JobManager" | "StorageRunner" | "AgentTurnRunner" | "RepoSession" | "PackageServiceInstance"; } interface Env { OAUTH_KV: KVNamespace; @@ -29,6 +29,7 @@ declare namespace Cloudflare { STORAGE_RUNNER: DurableObjectNamespace; AGENT_TURN_RUNNER: DurableObjectNamespace; REPO_SESSION: DurableObjectNamespace; + PACKAGE_SERVICE_INSTANCE: DurableObjectNamespace; } } interface Env extends Cloudflare.Env {} diff --git a/packages/worker/wrangler.jsonc b/packages/worker/wrangler.jsonc index 25c7c727bd..1afd08b0ff 100644 --- a/packages/worker/wrangler.jsonc +++ b/packages/worker/wrangler.jsonc @@ -50,6 +50,10 @@ "new_sqlite_classes": ["PackageRealtimeSession"], "tag": "v13", }, + { + "new_sqlite_classes": ["PackageServiceInstance"], + "tag": "v14", + }, ], "observability": { @@ -112,6 +116,10 @@ "class_name": "PackageRealtimeSession", "name": "PACKAGE_REALTIME_SESSION", }, + { + "class_name": "PackageServiceInstance", + "name": "PACKAGE_SERVICE_INSTANCE", + }, ], }, "d1_databases": [ @@ -197,6 +205,10 @@ "class_name": "PackageRealtimeSession", "name": "PACKAGE_REALTIME_SESSION", }, + { + "class_name": "PackageServiceInstance", + "name": "PACKAGE_SERVICE_INSTANCE", + }, ], }, "d1_databases": [ @@ -282,6 +294,10 @@ "class_name": "PackageRealtimeSession", "name": "PACKAGE_REALTIME_SESSION", }, + { + "class_name": "PackageServiceInstance", + "name": "PACKAGE_SERVICE_INSTANCE", + }, ], }, "d1_databases": [ diff --git a/tools/mcp-test-support.ts b/tools/mcp-test-support.ts index 2d9bf0cc2b..97cb61cefb 100644 --- a/tools/mcp-test-support.ts +++ b/tools/mcp-test-support.ts @@ -18,6 +18,7 @@ const primaryUserEmail = 'me@kentcdodds.com' const testUserPassword = 'secret' const localhost = '127.0.0.1' const defaultWaitTimeoutMs = process.env.CI ? 60_000 : 45_000 +const maxPortBindRetries = 5 type TestUser = { email: string @@ -57,99 +58,124 @@ export async function createTestDatabase() { export async function startDevServer(persistDir: string) { await applyMigrations(persistDir) - const mockCloudflarePort = await getPort({ host: localhost }) - const mockCloudflareOrigin = `http://${localhost}:${mockCloudflarePort}` - const mockCloudflareToken = `mock-cloudflare-${crypto.randomUUID()}` - const mockCloudflareProc = spawnProcess({ - cmd: [ - nodeBin, - '--env-file=packages/worker/.env', - './wrangler-env.ts', - 'dev', - '--local', - '--config', - 'packages/mock-servers/cloudflare/wrangler.jsonc', - '--port', - String(mockCloudflarePort), - '--ip', - localhost, - '--show-interactive-dev-session=false', - '--log-level', - 'error', - '--var', - `MOCK_API_TOKEN:${mockCloudflareToken}`, - ], - cwd: projectRoot, - env: { - ...process.env, - CLOUDFLARE_ENV: 'test', - }, - }) - const getMockCloudflareStdout = captureOutput(mockCloudflareProc.stdout) - const getMockCloudflareStderr = captureOutput(mockCloudflareProc.stderr) - - try { - await waitForUrlReady( - new URL('/__mocks/meta', mockCloudflareOrigin), - mockCloudflareProc.exited, - getMockCloudflareStdout, - getMockCloudflareStderr, - ) - } catch (error) { - await stopProcess(mockCloudflareProc).catch(() => undefined) - throw error - } - const port = await getPort({ host: localhost }) - const origin = `http://${localhost}:${port}` - const proc = spawnProcess({ - cmd: [ - nodeBin, - '--env-file=packages/worker/.env', - './wrangler-env.ts', - 'dev', - '--local', - '--persist-to', - persistDir, - '--port', - String(port), - '--ip', - localhost, - '--show-interactive-dev-session=false', - '--log-level', - 'error', - '--var', - `CLOUDFLARE_API_BASE_URL:${mockCloudflareOrigin}`, - '--var', - `CLOUDFLARE_API_TOKEN:${mockCloudflareToken}`, - '--var', - 'CLOUDFLARE_ACCOUNT_ID:cf_account_mock_123', - ], - cwd: projectRoot, - env: { - ...process.env, - CLOUDFLARE_ENV: 'test', - }, - }) - const getStdout = captureOutput(proc.stdout) - const getStderr = captureOutput(proc.stderr) + for (let attempt = 1; attempt <= maxPortBindRetries; attempt++) { + const mockCloudflarePort = await getPort({ host: localhost }) + const mockCloudflareOrigin = `http://${localhost}:${mockCloudflarePort}` + const mockCloudflareToken = `mock-cloudflare-${crypto.randomUUID()}` + const mockCloudflareProc = spawnProcess({ + cmd: [ + nodeBin, + '--env-file=packages/worker/.env', + './wrangler-env.ts', + 'dev', + '--local', + '--config', + 'packages/mock-servers/cloudflare/wrangler.jsonc', + '--port', + String(mockCloudflarePort), + '--ip', + localhost, + '--show-interactive-dev-session=false', + '--log-level', + 'error', + '--var', + `MOCK_API_TOKEN:${mockCloudflareToken}`, + ], + cwd: projectRoot, + env: { + ...process.env, + CLOUDFLARE_ENV: 'test', + }, + }) + const getMockCloudflareStdout = captureOutput(mockCloudflareProc.stdout) + const getMockCloudflareStderr = captureOutput(mockCloudflareProc.stderr) - try { - await waitForServerReady(origin, proc.exited, getStdout, getStderr) - } catch (error) { - await stopProcess(mockCloudflareProc).catch(() => undefined) - await stopProcess(proc).catch(() => undefined) - throw error - } + try { + await waitForUrlReady( + new URL('/__mocks/meta', mockCloudflareOrigin), + mockCloudflareProc.exited, + getMockCloudflareStdout, + getMockCloudflareStderr, + ) + } catch (error) { + await stopProcess(mockCloudflareProc).catch(() => undefined) + if (isPortAlreadyInUseError(error) && attempt < maxPortBindRetries) { + continue + } + throw error + } - return { - origin, - async [Symbol.asyncDispose]() { - await Promise.allSettled([ - stopProcess(proc), - stopProcess(mockCloudflareProc), - ]) - }, + const port = await getPort({ host: localhost }) + const origin = `http://${localhost}:${port}` + const proc = spawnProcess({ + cmd: [ + nodeBin, + '--env-file=packages/worker/.env', + './wrangler-env.ts', + 'dev', + '--local', + '--persist-to', + persistDir, + '--port', + String(port), + '--ip', + localhost, + '--show-interactive-dev-session=false', + '--log-level', + 'error', + '--var', + `CLOUDFLARE_API_BASE_URL:${mockCloudflareOrigin}`, + '--var', + `CLOUDFLARE_API_TOKEN:${mockCloudflareToken}`, + '--var', + 'CLOUDFLARE_ACCOUNT_ID:cf_account_mock_123', + ], + cwd: projectRoot, + env: { + ...process.env, + CLOUDFLARE_ENV: 'test', + }, + }) + const getStdout = captureOutput(proc.stdout) + const getStderr = captureOutput(proc.stderr) + + try { + await waitForServerReady(origin, proc.exited, getStdout, getStderr) + return { + origin, + async [Symbol.asyncDispose]() { + await Promise.allSettled([ + stopProcess(proc), + stopProcess(mockCloudflareProc), + ]) + }, + } + } catch (error) { + await stopProcess(mockCloudflareProc).catch(() => undefined) + await stopProcess(proc).catch(() => undefined) + if (isPortAlreadyInUseError(error) && attempt < maxPortBindRetries) { + continue + } + throw error + } } + + throw new Error('Failed to start MCP test dev servers after multiple retries.') +} + +function isPortAlreadyInUseError(error: unknown) { + if (!(error instanceof Error)) return false + const message = error.message.toLowerCase() + const codeValue = + typeof error === 'object' && error !== null + ? Reflect.get(error, 'code') + : undefined + const code = typeof codeValue === 'string' ? codeValue.toLowerCase() : '' + return ( + message.includes('address already in use') || + message.includes('eaddrinuse') || + code === 'eaddrinuse' + ) } export async function loginToApp(origin: string, user: TestUser) {