Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 17 additions & 63 deletions apps/chat-bot/src/relay/run.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,13 +7,7 @@
* the database, and none of it belongs on the path Cloudflare evaluates when it validates the
* uploaded script.
*/
import {
ConnectorCredentials,
WORKSPACE_CREDENTIALS,
type ChatConnector,
type ConnectorConfig,
type InboundEvent,
} from "@maple/chat-platform"
import type { ChatConnector, ConnectorCredentials, InboundEvent } from "@maple/chat-platform"
import { connectors } from "@maple/chat-platform/connectors"
import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare"
import { chatSessionStub } from "@maple/domain/chat-session-stub"
Expand All @@ -36,7 +30,7 @@ import {
resolveChatWorkspace,
} from "@maple/backend/services/integrations/chat-workspace-rows"
import { resolveConnectorConfig } from "../config.ts"
import { relayInboundEvent, WorkspaceLookupFailed, type RelayPorts, type RelayWorkspace } from "./turn.ts"
import { relayWithWorkspace, WorkspaceLookupFailed, type RelayPorts, type ResolvedWorkspace } from "./turn.ts"

/**
* This Worker's own SDK instance, at module scope so its buffers are the isolate's.
Expand Down Expand Up @@ -186,46 +180,14 @@ const lookupWorkspace = (
).pipe(lookupFailed(connectorId, "The chat workspace could not be read")),
)

/** What the one lookup answers: the relay's half, and the transport's. */
interface ResolvedWorkspace {
readonly relay: RelayWorkspace
readonly credentials: string | undefined
}

/**
* The config the connector's transport is built from: what the deployment set, plus this
* workspace's own credential where it stored one.
*/
const connectorCredentials = (
config: ConnectorConfig,
resolved: Option.Option<ResolvedWorkspace>,
): ConnectorConfig => {
if (Option.isNone(resolved) || resolved.value.credentials === undefined) return config
return new Map(config).set(WORKSPACE_CREDENTIALS, resolved.value.credentials)
}

const ports = (
host: RelayHost,
connector: ChatConnector<HttpClient.HttpClient | ConnectorCredentials>,
lookup: (
connectorId: ChatConnectorId,
workspaceId: string,
externalUserId: string | undefined,
) => Effect.Effect<Option.Option<ResolvedWorkspace>, WorkspaceLookupFailed>,
resolveWorkspace: RelayPorts["resolveWorkspace"],
): RelayPorts<HttpClient.HttpClient | ConnectorCredentials> => ({
outbound: connector.outbound,
supportsIdentity: connector.identity !== undefined,
/**
* The workspace, and — when a caller names a chat account — the Maple user it is linked to, in
* ONE connection.
*
* Sequential rather than parallel because the second question needs the first one's answer: a
* link is per org, and the org is what the workspace names.
*/
resolveWorkspace: (connectorId, workspaceId, externalUserId) =>
Effect.map(lookup(connectorId, workspaceId, externalUserId), (found) =>
Option.map(found, (workspace) => workspace.relay),
),
resolveWorkspace,
forgetWorkspace: (connectorId, workspaceId) =>
withDatabase(
host.env,
Expand Down Expand Up @@ -273,28 +235,20 @@ export const runInboundEvent = async (host: RelayHost, event: InboundEvent): Pro
const config = resolveConnectorConfig(host.env, connector)
if (config._tag === "missing") return

// One read for the whole event, memoized: the relay asks which org this is, and the transport is
// built from the credential the same row carries. `Effect.cached` is what keeps that one
// Postgres connection rather than two, without hoisting the read ahead of the relay's own
// decision about whether this message is a turn at all.
await Effect.runPromise(
Effect.gen(function* () {
const lookup = yield* Effect.cached(
lookupWorkspace(
host,
connector,
event.connector,
event.workspaceId,
// Only a click names a person; a message is answered for the workspace.
event.type === "action" ? event.actor.id : undefined,
),
)
const resolved = yield* Effect.orElseSucceed(lookup, () => Option.none<ResolvedWorkspace>())
yield* relayInboundEvent(
event,
ports(host, connector, () => lookup),
).pipe(Effect.provideService(ConnectorCredentials, connectorCredentials(config.config, resolved)))
}).pipe(
relayWithWorkspace(
event,
config.config,
lookupWorkspace(
host,
connector,
event.connector,
event.workspaceId,
// Only a click names a person; a message is answered for the workspace.
event.type === "action" ? event.actor.id : undefined,
),
(resolveWorkspace) => ports(host, connector, resolveWorkspace),
).pipe(
// oxlint-disable-next-line effecttsgo/strict-effect-provide
Effect.provide(Layer.mergeAll(FetchHttpClient.layer, workerEnvLayer(host.env), telemetry.layer)),
// On the fiber rather than in a `finally`: the flush is what exports this event's spans,
Expand Down
115 changes: 114 additions & 1 deletion apps/chat-bot/src/relay/turn.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import {
chatActionControlId,
chatConnectorId,
ChatOutboundError,
ConnectorCredentials,
encodeChatActionToken,
type ChatBlock,
type ChatChartRef,
Expand All @@ -33,13 +34,15 @@ import {
type ChatOutbound,
type ChatTarget,
type InboundAction,
type InboundEvent,
type InboundMessage,
type InboundWorkspaceRemoved,
WORKSPACE_CREDENTIALS,
} from "@maple/chat-platform"
import { Context, Duration, Effect, Logger, Option, References, Schema, Tracer } from "effect"
import { TestClock } from "effect/testing"
import { ConversationNotRecorded, FOLLOW_UP_WINDOW_MS } from "./conversation.ts"
import { relayInboundEvent, WorkspaceLookupFailed, type RelayPorts } from "./turn.ts"
import { relayInboundEvent, relayWithWorkspace, WorkspaceLookupFailed, type RelayPorts } from "./turn.ts"

const TESTCHAT = chatConnectorId("testchat")
const ORG = Schema.decodeSync(OrgId)("org_1")
Expand Down Expand Up @@ -1289,3 +1292,113 @@ describe("relaying everything else a connector reports", () => {
}),
)
})

describe("reading the workspace row", () => {
const NOW = Date.parse("2026-09-23T12:00:00.000Z")
const followUp: InboundMessage = {
...mention,
messageId: "message-2",
channelId: CONVERSATION,
mentionsBot: false,
}
const answered: ReadonlyArray<ChatMessage> = [
{
id: "a1",
role: "assistant",
text: "Checkout is slow.",
toolCalls: [],
createdAt: NOW - 60_000,
startSeq: 2,
},
]

/**
* The event relayed the way `run.ts` relays it: one counted read (the database connection and
* the decrypt), behind a connector that posts with the credential that read carries.
*/
const relay = (
event: InboundEvent,
stub: ChatSessionStub,
options?: { readonly opened?: ReadonlyArray<string>; readonly linked?: boolean },
) =>
Effect.gen(function* () {
const platform = chat()
const reads = { count: 0 }
const postedWith: Array<string | undefined> = []
const outbound: ChatOutbound<ConnectorCredentials> = {
...platform.outbound,
transport: Effect.gen(function* () {
const credentials = yield* ConnectorCredentials
const inner = yield* platform.outbound.transport
const using = <A, E>(effect: Effect.Effect<A, E>) =>
Effect.flatMap(credentials, (config) => {
postedWith.push(config.get(WORKSPACE_CREDENTIALS))
return effect
})
return {
...inner,
post: (target, blocks) => using(inner.post(target, blocks)),
edit: (ref, blocks) => using(inner.edit(ref, blocks)),
}
}),
}
yield* relayWithWorkspace(
event,
new Map(),
Effect.sync(() => {
reads.count += 1
return options?.linked === false
? Option.none()
: Option.some({ relay: { orgId: ORG }, credentials: "sealed" })
}),
(resolveWorkspace) => ({
...host(platform.outbound, stub, { opened: options?.opened ?? [] }).ports,
outbound,
resolveWorkspace,
}),
)
return { reads: reads.count, postedWith }
})

it.effect("reads nothing for a message it will not answer", () =>
Effect.gen(function* () {
const result = yield* relay(followUp, session([silentTurn], { transcript: answered }).stub)
expect(result).toEqual({ reads: 0, postedWith: [] })
}),
)

it.effect("reads it once for a follow-up in a conversation the bot owns", () =>
Effect.gen(function* () {
yield* TestClock.setTime(NOW)
const agent = session([silentTurn], { transcript: answered })
const result = yield* relay(followUp, agent.stub, { opened: [CONVERSATION] })
expect(agent.turns).toHaveLength(1)
expect(result.reads).toBe(1)
}),
)

it.effect("reads it once for a mention, and posts with the credential it carried", () =>
Effect.gen(function* () {
const result = yield* relay(mention, session([silentTurn]).stub)
expect(result.reads).toBe(1)
expect(result.postedWith.length).toBeGreaterThan(0)
expect(new Set(result.postedWith)).toEqual(new Set(["sealed"]))
}),
)

it.effect("posts with the deployment's config alone for a workspace nobody linked", () =>
Effect.gen(function* () {
const result = yield* relay(mention, session([silentTurn]).stub, { linked: false })
expect(result).toEqual({ reads: 1, postedWith: [undefined] })
}),
)

it.effect("reads it once for a click, and edits with the credential it carried", () =>
Effect.gen(function* () {
const agent = session([], { transcript: decidedTranscript("Approved by Ada.\nDone.") })
const result = yield* relay(click(), agent.stub)
expect(agent.settlements).toHaveLength(1)
expect(result).toEqual({ reads: 1, postedWith: ["sealed"] })
}),
)
})
45 changes: 45 additions & 0 deletions apps/chat-bot/src/relay/turn.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,15 +12,18 @@
* connector.
*/
import {
ConnectorCredentials,
decodeChatActionControlId,
driveChatTurn,
WORKSPACE_CREDENTIALS,
type ChatActionRequest,
type ChatChartRef,
type ChatConversation,
type ChatHistoryMessage,
type ChatOutbound,
type ChatOutboundTransport,
type ChatTarget,
type ConnectorConfig,
type InboundAction,
type InboundEvent,
type InboundMessage,
Expand Down Expand Up @@ -567,3 +570,45 @@ export const relayInboundEvent = <R>(
),
),
)

/** What the one workspace read answers: the relay's half, and the transport's credential. */
export interface ResolvedWorkspace {
readonly relay: RelayWorkspace
readonly credentials: string | undefined
}

/**
* {@link relayInboundEvent}, with the workspace row read at most once — and only when asked for.
*
* The relay's `resolveWorkspace` and the transport's `ConnectorCredentials` share one memoized
* read, so a message the gates turn away opens no connection and decrypts nothing, and one that
* proceeds costs exactly one of each.
*/
export const relayWithWorkspace = <R>(
event: InboundEvent,
config: ConnectorConfig,
lookup: Effect.Effect<Option.Option<ResolvedWorkspace>, WorkspaceLookupFailed>,
ports: (resolveWorkspace: RelayPorts<R>["resolveWorkspace"]) => RelayPorts<R>,
): Effect.Effect<void, never, Exclude<R, ConnectorCredentials>> =>
Effect.flatMap(Effect.cached(lookup), (read) =>
relayInboundEvent(
event,
ports(() =>
Effect.map(
read,
Option.map((found) => found.relay),
),
),
).pipe(
Effect.provideService(
ConnectorCredentials,
// A read that failed posts with the deployment's config alone; the relay has already
// refused to answer on it.
Effect.map(Effect.orElseSucceed(read, Option.none), (found) =>
Option.isNone(found) || found.value.credentials === undefined
? config
: new Map(config).set(WORKSPACE_CREDENTIALS, found.value.credentials),
),
),
),
)
3 changes: 2 additions & 1 deletion packages/chat-platform/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,8 @@ ChatEvent stream ──▶ makeChatTranscript ──▶ renderChatMessage ──
its token from `workspaceId` inside its own transport — the contract carries the address, never
the secret. The credential itself arrives as `ConnectorCredentials`, the configuration the host
resolved from the names the connector declared, so a host can drive any connector's outbound half
with two services it knows: that one and an HTTP client.
with two services it knows: that one and an HTTP client. It is an Effect the transport runs per
call, not at acquisition, so an event the host never answers never pays for resolving it.
- `src/driver.ts` — one turn, named by the assistant message id `beginTurn` answered with: placeholder,
throttled edits, retraction, splitting, approvals, and a short notice for a turn that did not
simply finish. A stream from seq 0 replays whole earlier turns, which is why the turn is named
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ const stub = (attempts: ReadonlyArray<Attempt>) => {
seen,
layer: Layer.mergeAll(
Layer.succeed(HttpClient.HttpClient)(client),
Layer.succeed(ConnectorCredentials)(new Map([[BOT_TOKEN_CONFIG, "bot-token"]])),
Layer.succeed(ConnectorCredentials)(Effect.succeed(new Map([[BOT_TOKEN_CONFIG, "bot-token"]]))),
),
}
}
Expand Down
Loading
Loading