[Graphite MQ] Draft PR GROUP:spec_875ba8 (PRs 4206, 4233, 4209, 4229, 4230, 4231)

[graphite-app]

This draft PR was created by the Graphite merge queue.
Trunk will be fast forwarded to the HEAD of this PR when CI passes, and the original PRs will be closed.

The following PRs are included in this draft PR:

• #4206
• #4233
• #4209
• #4229
• #4230
• #4231

[railway-app]

This PR was not deployed automatically as @graphite-app[bot] does not have access to the Railway project.

In order to get automatic PR deploys, please add @graphite-app[bot] to your workspace on Railway.

[claude]

Code Review

This is a large batch PR grouping 6 individual PRs (#4206, #4233, #4209, #4229, #4230, #4231). The changes introduce a typed schema system for events and queues, redesign the queue API, and make several ergonomic improvements. Overall the architecture is well-considered. Below are specific findings.

---

Potential Bug: isQueueSchemaDefinition can misidentify StandardSchemaV1 objects

File: rivetkit-typescript/packages/rivetkit/src/actor/schema.ts

export function isQueueSchemaDefinition(value: unknown): value is QueueSchemaDefinition {
    return (
        typeof value === "object" &&
        value !== null &&
        "message" in value &&
        (value as { message?: unknown }).message !== undefined
    );
}

A StandardSchemaV1 that describes an object with a message output field (e.g., a Zod schema for { message: string }) will pass this check and be mistakenly treated as a QueueSchemaDefinition. getValidationSchema would then try to use schema.message (the Zod field schema) as the actual validator.

Suggest adding a ~standard absence check:

"~standard" in value === false &&

---

Phantom type fields could unintentionally satisfy structural checks

File: rivetkit-typescript/packages/rivetkit/src/actor/schema.ts

export interface EventTypeToken<T> {
    readonly _eventType?: T;
}

Using optional underscore-prefixed fields for phantom types means any object that happens to have _eventType?: SomeType will satisfy EventTypeToken<SomeType> structurally. This makes the branding weak. Consider using a unique symbol key instead:

declare const _eventType: unique symbol;
export interface EventTypeToken<T> {
    readonly [_eventType]?: T;
}

This also avoids any risk of a user's data object accidentally satisfying the token interface.

---

validateSchemaSync throws Unsupported when given an async schema validator

File: rivetkit-typescript/packages/rivetkit/src/actor/schema.ts

export function validateSchemaSync<T extends AnySchemaConfig>(
    ...
): ValidationResult<...> {
    ...
    if (isStandardSchema(schema)) {
        const result = schema["~standard"].validate(data);
        if (isPromiseLike(result)) {
            throw new Unsupported("async schema validation");
        }
        ...
    }
}

When a user configures an async Standard Schema validator (e.g., one that performs database lookups) and calls conn.send(), they get an Unsupported error with a generic message. The error doesn't tell them which event/queue triggered this or that they need a synchronous validator. Consider surfacing the event name and a clearer message:

Async schema validation is not supported for event 'newMessage'. Use a synchronous validator (e.g., Zod's sync .parse()).

---

iter() with completable: true has no backpressure for missed completions

File: rivetkit-typescript/packages/rivetkit/src/actor/instance/queue.ts

async *iter<...>(opts?: QueueIterOptions<...>) {
    ...
    while (!this.#abortSignal.aborted) {
        const messages = resolvedOpts.completable === true
            ? await this.next<TName, true>({ ...completable: true })
            : ...;
        yield messages[0];
    }
}

If a user iterates with completable: true but forgets to call msg.complete() before the loop body exits, the next iteration will throw QueuePreviousMessageNotCompleted. While the error message is clear, this is an easy mistake in for await loops. Consider either:

• Documenting this prominently in the QueueIterOptions interface, or
• Surfacing it as a compile-time lint (e.g., complete returning a CompleteHandle that must be awaited)

The current runtime guard is functional, but users may be surprised.

---

Breaking changes warrant a version bump check

The following are breaking changes in this PR:

• QueueMessage fields removed: failureCount, availableAt, inFlight, inFlightAt
• c.queue.next("name") positional string argument replaced with opts object pattern
• connection.queue.message.send() client API replaced by connection.send("name", payload)
• QueueCompletionResult type removed
• QueueCompleteNotAllowed and QueueMessagePending error classes removed

Verify the package version has been (or will be) bumped to a minor/major version before publishing, depending on semver policy.

---

Minor: event() and queue() factory functions accept unused variadic args

File: rivetkit-typescript/packages/rivetkit/src/actor/schema.ts

export function event<T>(..._args: unknown[]): EventTypeToken<T> {
    return {} as EventTypeToken<T>;
}

The ..._args parameter exists (presumably for future use) but the leading _ communicates it's intentionally unused. If this is a forward-compatibility placeholder for passing a schema validator directly (e.g., event<T>(z.object(...))), it would be worth a brief comment explaining the intent.

---

Positive observations

• The for await...of c.queue.iter() API is a significant ergonomic improvement over the previous while(true) { const q = await c.queue.next(...) } pattern.
• The c.aborted shorthand is a clean improvement across the board.
• Using Standard Schema (@standard-schema/spec) for schema validation is a good choice—it ensures compatibility with Zod, Valibot, ArkType, and other validators without locking into one library.
• The SQLite runtime fallback chain (bun:sqlite → node:sqlite → better-sqlite3) improves portability across different runtimes.
• The RUN_FUNCTION_CONFIG_SYMBOL mechanism for attaching metadata to plain run functions without requiring a full RunConfig wrapper is a clean design.
• The collaborative-document bug fix (removeAwarenessStates as a standalone function rather than an instance method) is correct.