Merge pull request #121 from AgentWorkforce/feat/real-relay-sdk

feat: run agents against the real @agent-relay/agent SDK (Unit D)

Author: khaliqgant

agents/manual-chatbot/agent.ts

@@ -31,7 +31,7 @@ export default agent({
   async onEvent(ctx: Context, event: AgentEvent) {
     if (event.type !== "relaycast.message") return;
 
-    const msg = (await event.expand("full")) as MessageExpansion;
+    const msg = (await event.expand("full")) as unknown as MessageExpansion;
     const question = msg.data.text;
 
     // 1. Cheap intent classification: is this about proactive agents?

agents/notion-to-blog/agent.ts

@@ -183,10 +183,13 @@ export default agent({
 
     // ── 7. Open PR (idempotent — skips if one already exists) ───────────
     const octokit = await octokitFor(e);
+    // `ctx.once` returns undefined when this page revision was already
+    // published (deduped) — skip the Notion update + log on a repeat delivery.
     const prUrl = await ctx.once(
       `notion-pr:${pageId}:${page.last_edited_time}`,
       () => openPr(octokit, slug, frontmatter.title, mdx, page.url ?? ""),
     );
+    if (!prUrl) return;
 
     // ── 8. Update Notion: Status → Published, Published URL → PR URL ────
     await client.request("PATCH", `/pages/${pageId}`, {

agents/pr-reviewer/agent.ts

@@ -32,7 +32,7 @@ export default agent({
       return;
     }
 
-    const pr = (await event.expand("full")) as PrExpansion;
+    const pr = (await event.expand("full")) as unknown as PrExpansion;
     const mdxChanged = pr.data.changed_files.filter((f) => f.filename.endsWith(".mdx"));
     const codeChanged = pr.data.changed_files.filter(
       (f) => f.filename.startsWith("app/") || f.filename.startsWith("components/"),

agents/shared/log.ts

@@ -38,7 +38,7 @@ export async function writeLogEntry(ctx: Context, input: LogInput): Promise<Agen
   const next = [entry, ...without];
 
   await ctx.files.write(LOG_VFS_PATH, next, {
-    message: `[${entry.agent}] ${entry.action}`,
+    semantics: { commitMessage: `[${entry.agent}] ${entry.action}` },
   });
 
   return entry;

agents/shared/runtime/cloudflare-context.ts

@@ -46,6 +46,9 @@ export async function makeCloudflareContext(args: {
   const octokit = await getOctokit(env);
 
   const logger: Logger = {
+    debug(msg, meta) {
+      console.debug(`[${agentId}] ${msg}`, meta ?? "");
+    },
     info(msg, meta) {
       console.log(`[${agentId}] ${msg}`, meta ?? "");
     },
@@ -63,6 +66,21 @@ export async function makeCloudflareContext(args: {
     logger,
     signal,
 
+    // Burn tagging is a no-op here: this adapter talks to GitHub/Notion
+    // directly rather than through the metered relay primitives.
+    tagged: <T>(value: T): T => value,
+
+    // The relay primitive clients (relayfile/relaycron/relaycast) are not
+    // available in the Pages-Functions adapter — it uses GitHub/Notion APIs
+    // directly. No handler touches `ctx.raw`; fail loudly if one ever does.
+    raw: new Proxy({} as Context["raw"], {
+      get(_target, prop) {
+        throw new Error(
+          `ctx.raw.${String(prop)} is unavailable in the Pages-Functions context`,
+        );
+      },
+    }),
+
     files: {
       async read(path) {
         const repoPath = vfsToRepoPath(path);
@@ -72,12 +90,17 @@ export async function makeCloudflareContext(args: {
           path: repoPath,
           ref: REPO_BRANCH,
         });
-        return result ? { body: result.data, meta: { sha: result.sha } } : null;
+        // Map GitHub's content sha onto WorkspaceFile.revision (the SDK's
+        // relayfile revision channel) so writers can use it for compare-and-swap.
+        return result ? { path, body: result.data, revision: result.sha } : null;
       },
       async write(path, body, meta) {
         const repoPath = vfsToRepoPath(path);
-        const message =
-          (meta as { message?: string } | undefined)?.message ?? `[${agentId}] write ${path}`;
+        // WriteMeta has no free-form commit message; callers pass one via the
+        // SDK-sanctioned `semantics` extension point (`semantics.commitMessage`).
+        const commitMessage = (meta?.semantics as { commitMessage?: string } | undefined)
+          ?.commitMessage;
+        const message = commitMessage ?? `[${agentId}] write ${path}`;
         await writeRepoJson(octokit, {
           owner: REPO_OWNER,
           repo: REPO_NAME,

agents/shared/sdk.ts

@@ -1,133 +1,60 @@
 /**
- * Local mirror of the proactive-runtime SDK contract.
+ * The proactive-runtime SDK contract for these agents.
  *
- * Once `@agent-relay/agent` is published, replace this whole file with:
- *   export { agent, type AgentDefinition, type AgentEvent, type Context } from "@agent-relay/agent";
+ * The TYPES below are the published `@agent-relay/agent` contract — the package
+ * shipped, so the local mirror that used to live here is gone and handlers now
+ * type-check against the real SDK (`AgentDefinition`, `AgentEvent`, `Context`,
+ * `ScheduleSpec`).
  *
- * Until then we keep the types here so the handlers type-check, and `agent()`
- * runs as a no-op shim in local dev (registers the definition, doesn't dispatch).
+ * `agent()`, however, stays a LOCAL registrar rather than the SDK's hosted
+ * runtime. These agents dispatch from Cloudflare Pages Functions
+ * (`functions/api/...`) — request-scoped Workers that cannot hold the SDK's
+ * long-lived broker connection. The Pages Function reaches into
+ * `handle.definition.onEvent(ctx, event)` directly, so we expose `.definition`,
+ * which the hosted `agent()` (a connection-backed handle) does not. Move to the
+ * hosted runtime only if these ever run on a long-lived host instead of Pages
+ * Functions.
  */
-
+export type {
+  AgentDefinition,
+  AgentEvent,
+  AgentHandle,
+  Context,
+  EventType,
+  Logger,
+  ScheduleSpec
+} from "@agent-relay/agent";
+
+import type { AgentDefinition, AgentHandle, Context } from "@agent-relay/agent";
+
+/** Local trigger taxonomy used by the activity log; not part of the SDK. */
 export type Trigger = "time" | "change" | "message";
 
-export type ScheduleSpec =
-  | string
-  | { cron: string; tz?: string }
-  | { at: string | Date };
-
-export type EventType =
-  | "startup"
-  | "cron.tick"
-  | "relayfile.changed"
-  | "relaycast.message"
-  | (string & {}); // provider events: "github.pull_request.opened", "notion.page.updated", etc.
-
-export type AgentEvent<T extends EventType = EventType> = {
-  id: string;
-  workspace: string;
-  type: T;
-  occurredAt: string;
-  attempt: number;
-  resource: {
-    path: string;
-    kind: string;
-    id: string;
-    provider: string;
-  };
-  summary: {
-    title?: string;
-    status?: string;
-    priority?: string;
-    labels?: string[];
-    actor?: { id: string; displayName?: string };
-    fieldsChanged?: string[];
-    tags?: string[];
-  };
-  expand: <L extends "summary" | "full" | "diff" | "thread">(level?: L) => Promise<unknown>;
-  digest?: string;
-};
-
-export type Logger = {
-  info(msg: string, meta?: Record<string, unknown>): void;
-  warn(msg: string, meta?: Record<string, unknown>): void;
-  error(msg: string, meta?: Record<string, unknown>): void;
-};
-
-export type Context = {
-  workspace: string;
-  agentId: string;
-  logger: Logger;
-  signal: AbortSignal;
-  files: {
-    read(path: string): Promise<{ body: unknown; meta?: unknown } | null>;
-    write(path: string, body: unknown, meta?: Record<string, unknown>): Promise<void>;
-    delete(path: string): Promise<void>;
-    list(glob: string): Promise<{ path: string }[]>;
-  };
-  messages: {
-    post(channel: string, text: string, opts?: Record<string, unknown>): Promise<{ id: string }>;
-    reply(threadId: string, text: string, opts?: Record<string, unknown>): Promise<{ id: string }>;
-    dm(agentOrUser: string, text: string): Promise<{ id: string }>;
-  };
-  schedule: {
-    at(when: string | Date, payload?: unknown): Promise<{ id: string }>;
-    every(cron: string, payload?: unknown, opts?: { tz?: string }): Promise<{ id: string }>;
-    cancel(id: string): Promise<void>;
-  };
-  once<T>(key: string, fn: () => Promise<T>): Promise<T>;
-};
-
-export type AgentDefinition = {
-  workspace: string;
-  name?: string;
-  schedule?: ScheduleSpec | ScheduleSpec[];
-  watch?: string | string[];
-  inbox?: string | string[];
-  onEvent: (ctx: Context, event: AgentEvent) => Promise<void> | void;
-  onStart?: (ctx: Context) => Promise<void> | void;
-  onError?: (ctx: Context, error: Error, event: AgentEvent) => Promise<void> | void;
-  options?: {
-    concurrency?: number;
-    handlerTimeoutMs?: number;
-    replayOnStart?: "none" | string;
-  };
-};
-
-export type AgentHandle = {
-  ready: Promise<void>;
-  stop: () => Promise<void>;
-  trigger: (event: Partial<AgentEvent>) => Promise<void>;
-  ctx: Context;
-};
-
 /**
- * Local shim. When @agent-relay/agent ships, replace this whole module with:
- *   export { agent, type AgentDefinition, ... } from "@agent-relay/agent";
- *
- * Until then `agent()` returns a handle whose `trigger()` invokes the
- * registered `onEvent` synchronously with whatever event you pass it. This
- * is exactly the spec's "imperative trigger — useful in tests" semantic, and
- * it's how the Pages Function (functions/api/cron/[agent].ts) dispatches
- * cron.tick events until the real runtime takes over.
+ * Local handle shape. Extends the SDK {@link AgentHandle} with the original
+ * `definition` so the Pages Function dispatch can invoke `onEvent` directly
+ * (see the module comment).
  */
 export type AgentHandleWithDef = AgentHandle & {
   /** The original definition. Lets the Pages Function reach `onEvent`. */
   definition: AgentDefinition;
 };
 
+/**
+ * Register a proactive agent. In the Pages-Functions deployment this records
+ * the definition and returns a handle whose `definition` the entry-point
+ * dispatches against; it does not open a broker connection. `trigger()` throws
+ * — callers invoke `handle.definition.onEvent(ctx, event)` directly.
+ */
 export function agent(definition: AgentDefinition): AgentHandleWithDef {
-  // No real-time dispatch in the shim — schedules / watches / inbox are not
-  // wired to anything. The runtime takes that over. Until then, callers
-  // (e.g. functions/api/cron/[agent].ts) reach into `handle.definition` and
-  // invoke `onEvent(realCtx, event)` directly.
   return {
     definition,
     ready: Promise.resolve(),
     stop: async () => {},
     ctx: {} as Context,
     trigger: async () => {
       throw new Error(
-        "shim: handle.trigger() not implemented — invoke handle.definition.onEvent(ctx, event) directly",
+        "local registrar: invoke handle.definition.onEvent(ctx, event) directly",
       );
     },
   };

agents/weekly-digest/agent.ts

@@ -78,10 +78,14 @@ export default agent({
     const clusters = await clusterByTopic(ctx, fresh);
 
     // 4. Upsert the rolling issue.
-    const { issueUrl, issueNumber } = await ctx.once(
+    // `ctx.once` returns undefined when the keyed work was already done this
+    // period (deduped) — nothing more to do on a repeat tick.
+    const digestResult = await ctx.once(
       `digest:${weekKey(event.occurredAt)}`,
       () => upsertDigestIssue(ctx, clusters, event.occurredAt),
     );
+    if (!digestResult) return;
+    const { issueUrl, issueNumber } = digestResult;
 
     // 5. Persist seen URLs so next week dedupes against them.
     await ctx.files.write(

functions/api/cron/[agent].ts

@@ -8,14 +8,18 @@
  * look up the named agent, build a real Context, and invoke its `onEvent`
  * with a synthesised cron.tick event.
  *
- * When @agent-relay/agent ships, the runtime dispatches events directly
- * without going through HTTP and this whole file disappears.
+ * This HTTP entry point stays even though `@agent-relay/agent` has shipped:
+ * these agents run on Cloudflare Pages Functions (request-scoped Workers),
+ * which can't hold the SDK's long-lived broker connection, so we dispatch by
+ * calling `handle.definition.onEvent` directly. The event itself is built with
+ * the real SDK constructor (`createCronTickEvent`).
  */
+import { createCronTickEvent } from "@agent-relay/events";
 import weeklyDigest, { setEnv as setWeeklyDigestEnv } from "../../../agents/weekly-digest/agent";
 import notionToBlog, { setEnv as setNotionToBlogEnv } from "../../../agents/notion-to-blog/agent";
 import newsletterDrafter, { setEnv as setNewsletterDrafterEnv } from "../../../agents/newsletter-drafter/agent";
 import { makeCloudflareContext, type CfEnv } from "../../../agents/shared/runtime/cloudflare-context";
-import type { AgentEvent, AgentHandleWithDef } from "../../../agents/shared/sdk";
+import type { AgentHandleWithDef } from "../../../agents/shared/sdk";
 
 type Params = { agent: string };
 
@@ -67,21 +71,13 @@ export const onRequestPost: PagesFunction<CfEnv, "agent"> = async (ctx) => {
   });
 
   const occurredAt = new Date().toISOString();
-  const event: AgentEvent<"cron.tick"> = {
-    id: `cron-${occurredAt}-${agentName}`,
+  const event = createCronTickEvent({
     workspace: entry.workspace,
-    type: "cron.tick",
+    schedule: agentName,
+    id: `cron-${occurredAt}-${agentName}`,
     occurredAt,
-    attempt: 1,
-    resource: {
-      path: `/_internal/cron/${agentName}`,
-      kind: "cron.tick",
-      id: agentName,
-      provider: "internal",
-    },
     summary: { title: `${agentName} cron tick` },
-    expand: async () => ({}),
-  };
+  });
 
   try {
     await entry.handle.definition.onEvent(agentCtx, event);

functions/api/notion-webhook.ts

@@ -5,14 +5,15 @@
  * Route: POST /api/notion-webhook
  *
  * Until M2 data triggers ship, this endpoint receives requests and
- * synthesizes relayfile.changed events for the notion-to-blog agent.
- * When the runtime handles this natively, this file gets deleted.
+ * synthesizes relayfile.changed events (via the SDK's `createAgentEvent`) for
+ * the notion-to-blog agent. The dispatch stays HTTP because Pages Functions
+ * can't host the SDK's long-lived broker runtime.
  *
  * Also accepts GET for manual trigger: GET /api/notion-webhook?page_id=<id>&secret=<s>
  */
+import { createAgentEvent } from "@agent-relay/events";
 import notionToBlog, { setEnv as setNotionToBlogEnv } from "../../agents/notion-to-blog/agent";
 import { makeCloudflareContext, type CfEnv } from "../../agents/shared/runtime/cloudflare-context";
-import type { AgentEvent } from "../../agents/shared/sdk";
 import {
   NotionApiClient,
   serializePropertyMap,
@@ -73,24 +74,23 @@ async function dispatchForPage(env: CfEnv, pageId: string): Promise<Response> {
   });
 
   const occurredAt = new Date().toISOString();
-  const event: AgentEvent<"relayfile.changed"> = {
-    id: `notion-${pageId}-${occurredAt}`,
-    workspace: "proactive-agents",
-    type: "relayfile.changed",
-    occurredAt,
-    attempt: 1,
-    resource: {
-      path: `/notion/databases/drafts/pages/${pageId}`,
-      kind: "page",
-      id: pageId,
-      provider: "notion",
-    },
-    summary: {
-      title: propValue(props.Title),
-      status: propValue(props.Status)?.toLowerCase(),
+  const path = `/notion/databases/drafts/pages/${pageId}`;
+  const event = createAgentEvent(
+    {
+      id: `notion-${pageId}-${occurredAt}`,
+      workspace: "proactive-agents",
+      type: "relayfile.changed",
+      occurredAt,
+      path,
+      action: "updated",
+      resource: { path, kind: "page", id: pageId, provider: "notion" },
+      summary: {
+        title: propValue(props.Title),
+        status: propValue(props.Status)?.toLowerCase(),
+      },
     },
-    expand: async () => page,
-  };
+    { loadFull: async () => ({ level: "full", path, data: page as unknown as Record<string, unknown> }) }
+  );
 
   try {
     await notionToBlog.definition.onEvent(agentCtx, event);

memory/workspace/.relay/state.json

@@ -0,0 +1 @@
+{"workspaceId":"rw_7ccfea89","remoteRoot":"/memory/workspace","localRoot":"/home/daytona/workspace/memory/workspace","mode":"poll","syncMode":"mirror","intervalMs":5000,"lastReconcileAt":"2026-06-09T16:12:40.760983493Z","lastSuccessfulReconcileAt":"2026-06-09T16:12:40.760983493Z","staleAfter":"2026-06-09T16:12:50.760983493Z","status":"ready","states":{"stale":false,"offline":false,"hasConflicts":false,"hasPendingWriteback":false},"pendingWriteback":0,"pendingConflicts":0,"deniedPaths":0,"counters":{"snapshotDeleteBlocked":265},"circuit":{"open":false,"openedAt":"0001-01-01T00:00:00Z","windowMs":60000,"cooldownMs":30000,"threshold":5,"nextRetry":"0001-01-01T00:00:00Z"}}