feat(eve): add authenticated create-once sessions

.changeset/session-create-once.md

@@ -0,0 +1,5 @@
+---
+"eve": patch
+---
+
+Add authenticated create-once session requests through `operationId`. Concurrent or retried creates adopt the active session that first claimed the operation without dispatching duplicate input.

docs/channels/eve.mdx

@@ -41,6 +41,19 @@ curl -X POST https://<deployment>/eve/v1/session \
 # {"continuationToken":"eve:7f3c...","ok":true,"sessionId":"ses_01h..."}
 ```
 
+Authenticated callers that may retry a create request can pass their own `operationId` for
+create-once semantics. The same operation under the same authenticated principal returns the
+active session it already created instead of dispatching the input again. The first accepted
+payload wins; retries with different input still return that first session. Anonymous callers
+cannot use `operationId`, and operation ownership expires when the session is no longer resumable.
+
+```bash
+curl -X POST https://<deployment>/eve/v1/session \
+  -H "Authorization: Bearer <token>" \
+  -H "Content-Type: application/json" \
+  -d '{"message":"What is the weather in Paris?","operationId":"order-4213-research"}'
+```
+
 Stream that session's events as newline-delimited JSON (`application/x-ndjson; charset=utf-8`), one event object per line:
 
 ```bash

e2e/fixtures/agent-basic-runtime/agent/channels/eve.ts

@@ -0,0 +1,25 @@
+import type { AuthFn } from "eve/channels/auth";
+import { eveChannel } from "eve/channels/eve";
+import type { SessionAuthContext } from "eve/context";
+
+const PRINCIPAL_A = "Bearer e2e-create-once-a";
+const PRINCIPAL_B = "Bearer e2e-create-once-b";
+
+function principal(issuer: string): SessionAuthContext {
+  return {
+    attributes: {},
+    authenticator: "e2e-create-once",
+    issuer,
+    principalId: "shared-principal-id",
+    principalType: "user",
+    subject: "shared-subject",
+  };
+}
+
+const authenticateA: AuthFn<Request> = (request) =>
+  request.headers.get("authorization") === PRINCIPAL_A ? principal("issuer-a") : null;
+const authenticateB: AuthFn<Request> = (request) =>
+  request.headers.get("authorization") === PRINCIPAL_B ? principal("issuer-b") : null;
+const authenticateEvalDriver: AuthFn<Request> = () => principal("eval-driver");
+
+export default eveChannel({ auth: [authenticateA, authenticateB, authenticateEvalDriver] });

e2e/fixtures/agent-basic-runtime/evals/runtime/session-create-idempotency.eval.ts

@@ -0,0 +1,98 @@
+import { defineEval, type EveEvalTargetHandle } from "eve/evals";
+import { equals, satisfies } from "eve/evals/expect";
+
+interface CreateSessionResponse {
+  readonly continuationToken: string;
+  readonly ok: true;
+  readonly sessionId: string;
+}
+
+const PRINCIPAL_A = "Bearer e2e-create-once-a";
+const PRINCIPAL_B = "Bearer e2e-create-once-b";
+
+export default defineEval({
+  description:
+    "Concurrent and serial create retries dispatch once and remain isolated across issuers.",
+  async test(t) {
+    const operationId = `session-create-idempotency-${crypto.randomUUID()}`;
+    const message = `CREATE-ONCE-INITIAL-${crypto.randomUUID()}`;
+    const [first, concurrent] = await Promise.all([
+      createSession(t.target, PRINCIPAL_A, operationId, message),
+      createSession(t.target, PRINCIPAL_A, operationId, message),
+    ]);
+    const replay = await createSession(t.target, PRINCIPAL_A, operationId, message);
+    const otherIssuer = await createSession(t.target, PRINCIPAL_B, operationId, message);
+
+    for (const retry of [concurrent, replay]) {
+      await t.require(retry.sessionId, equals(first.sessionId));
+      await t.require(retry.continuationToken, equals(first.continuationToken));
+    }
+    await t.require(
+      otherIssuer,
+      satisfies(
+        (value: CreateSessionResponse) =>
+          value.sessionId !== first.sessionId &&
+          value.continuationToken !== first.continuationToken,
+        "the same operation under another issuer owns a distinct session",
+      ),
+    );
+
+    const [firstTurn, issuerTurn] = await Promise.all([
+      t.target.watchTurn(first.sessionId).result(),
+      t.target.watchTurn(otherIssuer.sessionId).result(),
+    ]);
+    firstTurn.expectOk();
+    firstTurn.event("message.received", { count: 1, data: { message } });
+    firstTurn.event("step.started", { count: 1 });
+    issuerTurn.expectOk();
+    issuerTurn.event("message.received", { count: 1, data: { message } });
+    issuerTurn.event("step.started", { count: 1 });
+
+    const probe = `CREATE-ONCE-PROBE-${crypto.randomUUID()}`;
+    const liveProbe = t.target.watchTurn(first.sessionId, { startIndex: firstTurn.events.length });
+    await continueSession(
+      t.target,
+      PRINCIPAL_A,
+      first.sessionId,
+      first.continuationToken,
+      probe,
+    );
+    const probeTurn = await liveProbe.result();
+    probeTurn.expectOk();
+    probeTurn.event("message.received", { count: 1, data: { message: probe } });
+    probeTurn.event("step.started", { count: 1 });
+  },
+});
+
+async function createSession(
+  target: EveEvalTargetHandle,
+  authorization: string,
+  operationId: string,
+  message: string,
+): Promise<CreateSessionResponse> {
+  const response = await target.fetch("/eve/v1/session", {
+    body: JSON.stringify({ message, operationId }),
+    headers: { authorization, "content-type": "application/json" },
+    method: "POST",
+  });
+  const text = await response.text();
+  if (!response.ok) throw new Error(`POST /eve/v1/session failed (${response.status}): ${text}`);
+  return JSON.parse(text) as CreateSessionResponse;
+}
+
+async function continueSession(
+  target: EveEvalTargetHandle,
+  authorization: string,
+  sessionId: string,
+  continuationToken: string,
+  message: string,
+): Promise<void> {
+  const response = await target.fetch(`/eve/v1/session/${encodeURIComponent(sessionId)}`, {
+    body: JSON.stringify({ continuationToken, message }),
+    headers: { authorization, "content-type": "application/json" },
+    method: "POST",
+  });
+  if (!response.ok) {
+    throw new Error(`POST continuation failed (${response.status}): ${await response.text()}`);
+  }
+}

packages/eve/src/channel/routes.ts

@@ -95,9 +95,11 @@ type BaseSendOptions = {
   continuationToken: string;
   /**
    * `"resume"` requires an active session and propagates a typed no-active-session
-   * error. `"resume-or-start"` preserves the default channel behavior.
+   * error. `"create-once"` adopts an existing or concurrently-created owner
+   * without delivering the duplicate input. `"resume-or-start"` preserves the
+   * default channel behavior.
    */
-  intent?: "resume" | "resume-or-start";
+  intent?: "create-once" | "resume" | "resume-or-start";
   /**
    * The original (top-level) caller's auth for a newly started session,
    * becoming `session.auth.initiator`. Defaults to {@link auth} when omitted

packages/eve/src/channel/send.test.ts

@@ -144,6 +144,50 @@ describe("createSendFn", () => {
     expect(runtime.dispatchContinuation).toHaveBeenCalledTimes(2);
   });
 
+  it("adopts a concurrent create-once winner without delivering duplicate input", async () => {
+    const runtime = createRuntime();
+    vi.mocked(runtime.createSession).mockRejectedValue(
+      new RuntimeSessionOwnershipConflictError({
+        continuationToken: "test:token",
+        ownerSessionId: "winner",
+        sessionId: "loser",
+      }),
+    );
+    vi.mocked(runtime.resolveContinuation).mockResolvedValue(undefined);
+
+    await expect(
+      createSendFn(
+        runtime,
+        ADAPTER,
+        "test",
+      )("hello", {
+        auth: null,
+        continuationToken: "token",
+        intent: "create-once",
+      }),
+    ).resolves.toMatchObject({ id: "winner" });
+    expect(runtime.dispatchContinuation).not.toHaveBeenCalled();
+  });
+
+  it("adopts an existing create-once owner without delivering duplicate input", async () => {
+    const runtime = createRuntime();
+    vi.mocked(runtime.resolveContinuation).mockResolvedValue({ sessionId: "winner" });
+
+    await expect(
+      createSendFn(
+        runtime,
+        ADAPTER,
+        "test",
+      )("hello", {
+        auth: null,
+        continuationToken: "token",
+        intent: "create-once",
+      }),
+    ).resolves.toMatchObject({ id: "winner" });
+    expect(runtime.dispatchContinuation).not.toHaveBeenCalled();
+    expect(runtime.createSession).not.toHaveBeenCalled();
+  });
+
   it("forwards the turn caller on the session command", async () => {
     const runtime = createRuntime({ sessionId: "existing-session-id", status: "accepted" });
     const caller = {

packages/eve/src/channel/send.ts

@@ -55,7 +55,11 @@ export function createSendFn<TState = undefined>(
         : undefined;
     };
 
-    const existing = await dispatch();
+    const existingOwner = async (): Promise<Session | undefined> => {
+      const owner = await runtime.resolveContinuation(continuationToken);
+      return owner === undefined ? undefined : createSession(owner.sessionId, rawToken, runtime);
+    };
+    const existing = intent === "create-once" ? await existingOwner() : await dispatch();
     if (existing !== undefined) return existing;
     if (intent === "resume") throw new RuntimeNoActiveSessionError(continuationToken);
 
@@ -91,6 +95,9 @@ export function createSendFn<TState = undefined>(
       return createSession(handle.sessionId, rawToken, runtime);
     } catch (error) {
       if (!isRuntimeSessionOwnershipConflictError(error)) throw error;
+      if (intent === "create-once") {
+        return createSession(error.ownerSessionId, rawToken, runtime);
+      }
       const winner = await dispatch();
       if (winner !== undefined) return winner;
       throw error;

packages/eve/src/execution/workflow-entry.test.ts

@@ -198,7 +198,7 @@ describe("workflowEntry", () => {
     expect(terminateChildSessionsStep).toHaveBeenCalledWith({ sessionState });
   });
 
-  it("fails a conflicting delivery hook before dispatching the first turn", async () => {
+  it("exits a conflicting initial continuation before dispatching the first turn", async () => {
     const sessionState = createBaseSessionState();
     const dispose = vi.fn();
     vi.mocked(createSessionStep).mockResolvedValue(createSessionStepResultForMock(sessionState));
@@ -218,25 +218,14 @@ describe("workflowEntry", () => {
         input: { message: "duplicate" },
         serializedContext: createSerializedContext(),
       }),
-    ).rejects.toMatchObject({
-      message: "Agent workflow failed. Inspect the private session trace for details.",
-      name: "EveWorkflowFailure",
-    });
+    ).resolves.toEqual({ output: "" });
 
-    expect(emitTerminalSessionFailureStep).toHaveBeenCalledWith(
-      expect.objectContaining({
-        error: expect.objectContaining({
-          conflictingRunId: "wrun_owner",
-          name: "HookConflictError",
-          token: "http:test",
-        }),
-      }),
-    );
+    expect(emitTerminalSessionFailureStep).not.toHaveBeenCalled();
     expect(dispatchTurnStep).not.toHaveBeenCalled();
     expect(dispose).toHaveBeenCalledOnce();
   });
 
-  it("normalizes the getConflict fallback error before dispatching the first turn", async () => {
+  it("also exits when a legacy world reports the initial continuation conflict", async () => {
     const sessionState = createBaseSessionState();
     const dispose = vi.fn();
     const fallbackError = Object.assign(new Error("legacy hook conflict"), {
@@ -262,20 +251,9 @@ describe("workflowEntry", () => {
         input: { message: "duplicate" },
         serializedContext: createSerializedContext(),
       }),
-    ).rejects.toMatchObject({
-      message: "Agent workflow failed. Inspect the private session trace for details.",
-      name: "EveWorkflowFailure",
-    });
+    ).resolves.toEqual({ output: "" });
 
-    expect(emitTerminalSessionFailureStep).toHaveBeenCalledWith(
-      expect.objectContaining({
-        error: expect.objectContaining({
-          message: 'Hook token "http:test" is already in use',
-          name: "HookConflictError",
-          token: "http:test",
-        }),
-      }),
-    );
+    expect(emitTerminalSessionFailureStep).not.toHaveBeenCalled();
     expect(dispatchTurnStep).not.toHaveBeenCalled();
     expect(dispose).toHaveBeenCalledOnce();
   });

packages/eve/src/execution/workflow-entry.ts

@@ -31,7 +31,7 @@ import { createSessionStep } from "#execution/create-session-step.js";
 import { settleCancelledTurnStep } from "#execution/settle-cancelled-turn-step.js";
 import { emitTerminalSessionFailureStep } from "#execution/terminal-session-failure-step.js";
 import { fireSessionCallbackStep } from "#execution/session-callback-step.js";
-import { disposeHook } from "#execution/hook-ownership.js";
+import { disposeHook, isHookConflictError } from "#execution/hook-ownership.js";
 import { createSessionCommandInbox } from "#execution/session-command-inbox.js";
 import { sessionCommandHookToken } from "#execution/session-command-token.js";
 import { DEFAULT_SESSION_TIMEOUT_MS } from "#execution/session-timeout.js";
@@ -344,7 +344,14 @@ async function runDriverLoop(input: {
 
   try {
     if (input.sessionState.continuationToken) {
-      await commandInbox.rekeyContinuation(input.sessionState.continuationToken);
+      try {
+        await commandInbox.rekeyContinuation(input.sessionState.continuationToken);
+      } catch (error) {
+        // A concurrent create may start multiple candidates before one owns
+        // the shared alias. The loser exits before dispatching its first turn.
+        if (!isHookConflictError(error)) throw error;
+        return { kind: "result", result: { output: "" } };
+      }
     }
     await sessionTimeout?.start();
 

packages/eve/src/execution/workflow-runtime.ts

@@ -179,7 +179,6 @@ export function createWorkflowRuntime(config: {
         throw error;
       }
 
-      await waitForOwnedCommandHook(sessionCommandHookToken(run.runId), run.runId);
       if (input.continuationToken) {
         const owner = await waitForCommandHookOwner(input.continuationToken);
         if (owner.runId !== run.runId) {
@@ -190,6 +189,7 @@ export function createWorkflowRuntime(config: {
           });
         }
       }
+      await waitForOwnedCommandHook(sessionCommandHookToken(run.runId), run.runId);
 
       let events: ReadableStream<MessageStreamEvent> | undefined;
       const getEvents = () => {