# Files changed · PR \#17129 · pingdotgg/t3code
[View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/17129/diffs)
Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code)
Visibility: public
[feat\(server,web,mobile\): an agent sets up a link to another environment](https://git.cafe/pingdotgg/t3code/pull/17129.md)
Version: 32
Head commit: 71bc8a87e79f3a40b808fe38dfe90a60caf556fb
Base commit: 70a2a7bdd25d1c3729a864517727914fc3b01d2b
Comparison base: 70a2a7bdd25d1c3729a864517727914fc3b01d2b
Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing.
## apps/mobile/src/features/threads/LinkRequestCard.tsx
Status: added; +514; −0
```diff
@@ -1,0 +1,514 @@
+import { useAtomValue } from "@effect/atom-react";
+import {
+ type LinkableMachine,
+ linkRequestAnswerInput,
+ linkRequestDefaultAccess,
+ linkRequestDisplay,
+ linkRequestFailureMessage,
+ linkRequestInitialMachine,
+ type LinkRequestItem,
+ linkRequestTargetName,
+ rankMachinesByHint,
+} from "@t3tools/client-runtime/link-request";
+import {
+ isAtomCommandInterrupted,
+ squashAtomCommandFailure,
+} from "@t3tools/client-runtime/state/runtime";
+import {
+ AuthAccessWriteScope,
+ type AuthMcpClientAccess,
+ type EnvironmentId,
+ type OrchestrationV2ProjectedTurnItem,
+} from "@t3tools/contracts";
+import * as Option from "effect/Option";
+import * as Result from "effect/Result";
+import { useMemo, useRef, useState } from "react";
+import { Modal, Platform, Pressable, ScrollView, View, type ColorValue } from "react-native";
+import { useSafeAreaInsets } from "react-native-safe-area-context";
+
+import { SymbolView, type AppSymbolName } from "../../components/ui/AppSymbol";
+import { AppText as Text, AppTextInput as TextInput } from "../../components/ui/AppText";
+import { EnvironmentMachineSymbol } from "../../components/EnvironmentMachineSymbol";
+import { useLinkableMachines } from "../../state/peerLinks";
+import { mintPairingCodeOn } from "../../state/peerPairingCode";
+import { serverEnvironment } from "../../state/server";
+import { useEnvironmentScope, usePreparedConnection } from "../../state/session";
+import { useAtomCommand } from "../../state/use-atom-command";
+import { SettingsChoiceRow } from "../settings/components/SettingsChoiceRow";
+import { SettingsSection } from "../settings/components/SettingsSection";
+import { Button } from "../../components/ui/Button";
+import { RUNTIME_MODE_CHOICES } from "./thread-settings-options";
+
+/** The same five levels as web's access picker: read-only, then the runtime modes. */
+const ACCESS_CHOICES: ReadonlyArray<{
+ readonly access: AuthMcpClientAccess;
+ readonly label: string;
+ readonly description: string;
+}> = [
+ {
+ access: "read-only",
+ label: "Read only",
+ description: "Read projects and threads. Cannot start, message or change anything.",
+ },
+ ...RUNTIME_MODE_CHOICES.map((choice) => ({ ...choice, access: choice.mode })),
+];
+const accessLabel = (access: AuthMcpClientAccess) =>
+ ACCESS_CHOICES.find((choice) => choice.access === access)?.label ?? access;
+
+const LINK_SYMBOL: AppSymbolName = { ios: "link", android: "desktop_windows" };
+
+/** The machine list's row for linking by an address typed in. */
+const ANOTHER_ADDRESS = "another-address";
+
+/** Whether the agent left the machine for the user to pick. */
+const picksMachine = (item: LinkRequestItem) =>
+ item.environmentId === undefined && item.url === undefined;
+
+/**
+ * Feed card for a link an agent asked the user to make. The user picks the
+ * access in a sheet; the pairing code lives only in that sheet's state and
+ * the RPC payload, and is never logged, alerted, or persisted.
+ */
+export function LinkRequestCard(props: {
+ readonly environmentId: EnvironmentId;
+ readonly projectedItem: OrchestrationV2ProjectedTurnItem;
+ readonly iconColor: ColorValue;
+}) {
+ const { item, visibility } = props.projectedItem;
+ if (item.type !== "link_request") return null;
+ const display = linkRequestDisplay(item, visibility, accessLabel);
+ if (display.kind === "pending") {
+ return ;
+ }
+ const icon: AppSymbolName =
+ display.kind === "pending-elsewhere"
+ ? LINK_SYMBOL
+ : display.outcome === "linked"
+ ? "checkmark"
+ : display.outcome === "failed"
+ ? "xmark"
+ : "minus";
+ return (
+
+
+
+ {display.label}
+
+
+ );
+}
+
+function PendingLinkRequest(props: {
+ readonly environmentId: EnvironmentId;
+ readonly item: LinkRequestItem;
+}) {
+ const { item } = props;
+ const [open, setOpen] = useState(false);
+ const [declining, setDeclining] = useState(false);
+ const [error, setError] = useState(null);
+ const answer = useAtomCommand(serverEnvironment.answerLinkRequest, {
+ label: "decline link request",
+ reportFailure: false,
+ reportDefect: false,
+ });
+ const canAnswer = useAtomValue(
+ serverEnvironment.answerLinkRequest.permissionAtom(props.environmentId),
+ );
+ const target = linkRequestTargetName(item);
+
+ const decline = async () => {
+ const input = linkRequestAnswerInput(item, { type: "decline" });
+ if (input === null) return;
+ setDeclining(true);
+ setError(null);
+ const result = await answer({ environmentId: props.environmentId, input }).finally(() =>
+ setDeclining(false),
+ );
+ if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) {
+ setError(linkRequestFailureMessage(squashAtomCommandFailure(result)));
+ }
+ };
+
+ return (
+
+
+
+ {picksMachine(item)
+ ? "Link a machine to this environment"
+ : `Link ${target} to this environment`}
+
+ {item.reason.trim() ? (
+ {item.reason}
+ ) : null}
+ {item.label !== undefined && item.url !== undefined ? (
+
+ {item.url}
+
+ ) : null}
+
+ {error !== null ? (
+
+ {error}
+
+ ) : null}
+ {canAnswer ? (
+ <>
+
+ );
+}
+
+/** What the sheet links; see web's LinkRequestDialog for the same rules. */
+type SheetTarget =
+ | { readonly kind: "none" }
+ | {
+ readonly kind: "machine";
+ readonly environmentId: EnvironmentId;
+ readonly label: string;
+ readonly urls: ReadonlyArray;
+ readonly linked: boolean;
+ }
+ | { readonly kind: "address" };
+
+function resolveTarget(
+ item: LinkRequestItem,
+ machines: ReadonlyArray,
+ choice: string | null,
+): SheetTarget {
+ if (item.environmentId === undefined && item.url !== undefined) return { kind: "address" };
+ const chosen = item.environmentId ?? choice;
+ if (chosen === null) return { kind: "none" };
+ if (chosen === ANOTHER_ADDRESS) return { kind: "address" };
+ const machine = machines.find((candidate) => candidate.environmentId === chosen);
+ return {
+ kind: "machine",
+ environmentId: machine?.environmentId ?? (chosen as EnvironmentId),
+ label: machine?.label ?? linkRequestTargetName(item),
+ urls: [...new Set([...(machine?.urls ?? []), ...(item.url === undefined ? [] : [item.url])])],
+ linked: machine?.linked ?? false,
+ };
+}
+
+function LinkRequestSheet(props: {
+ readonly environmentId: EnvironmentId;
+ readonly item: LinkRequestItem;
+ readonly onClose: () => void;
+}) {
+ const { item } = props;
+ const insets = useSafeAreaInsets();
+ const pickMachine = picksMachine(item);
+ const listed = useLinkableMachines(props.environmentId, true);
+ const machines = useMemo(() => rankMachinesByHint(listed, item.hint), [listed, item.hint]);
+ const [choice, setChoice] = useState(null);
+ const [address, setAddress] = useState(item.url ?? "");
+ const [access, setAccess] = useState(() => linkRequestDefaultAccess(item));
+ const [pairingCode, setPairingCode] = useState("");
+ const [busy, setBusy] = useState(false);
+ const [error, setError] = useState(null);
+ const [mintFailed, setMintFailed] = useState>(new Set());
+ // Submit then a tap can both run before a re-render; this guard is synchronous.
+ const inFlight = useRef(false);
+ const answer = useAtomCommand(serverEnvironment.answerLinkRequest, {
+ label: "answer link request",
+ // The failure cause holds the request, and with it the code; keep it out of the console.
+ reportFailure: false,
+ reportDefect: false,
+ });
+ const selected = choice ?? linkRequestInitialMachine(item, machines);
+ const target = resolveTarget(item, machines, selected);
+ const targetName = target.kind === "machine" ? target.label : linkRequestTargetName(item);
+ // This phone can mint a code there when it is connected to that environment
+ // with a session that may create pairing links.
+ const targetId = target.kind === "machine" ? target.environmentId : null;
+ const connected = usePreparedConnection(targetId);
+ const mayMint = useEnvironmentScope(targetId, AuthAccessWriteScope);
+ const canMint =
+ targetId !== null && Option.isSome(connected) && mayMint && !mintFailed.has(targetId);
+ const needsAddress =
+ target.kind === "address" || (target.kind === "machine" && target.urls.length === 0);
+ const useExisting = target.kind === "machine" && target.linked;
+ const ready =
+ target.kind !== "none" &&
+ (useExisting ||
+ ((!needsAddress || address.trim().length > 0) && (canMint || pairingCode.trim().length > 0)));
+
+ const prepareAnswer = async () => {
+ if (target.kind === "none") return null;
+ if (target.kind === "machine" && target.linked) {
+ return linkRequestAnswerInput(item, {
+ type: "use-existing",
+ environmentId: target.environmentId,
+ });
+ }
+ let code = pairingCode;
+ if (canMint && targetId !== null) {
+ const minted = await mintPairingCodeOn(targetId, access);
+ if (Result.isFailure(minted)) {
+ setMintFailed((failed) => new Set(failed).add(targetId));
+ setError(linkRequestFailureMessage(minted.failure));
+ return null;
+ }
+ code = minted.success;
+ }
+ return linkRequestAnswerInput(item, {
+ type: "link",
+ target:
+ target.kind === "address"
+ ? { url: address }
+ : { ...target, urls: needsAddress ? [address.trim()] : target.urls },
+ access,
+ pairingCode: code,
+ });
+ };
+
+ const submit = async () => {
+ if (inFlight.current || target.kind === "none") return;
+ inFlight.current = true;
+ setBusy(true);
+ setError(null);
+ const input = await prepareAnswer();
+ const result =
+ input === null ? null : await answer({ environmentId: props.environmentId, input });
+ inFlight.current = false;
+ setBusy(false);
+ if (result === null) return;
+ if (result._tag === "Success") {
+ setPairingCode("");
+ props.onClose();
+ } else if (!isAtomCommandInterrupted(result)) {
+ setError(linkRequestFailureMessage(squashAtomCommandFailure(result)));
+ }
+ };
+
+ return (
+ {
+ if (!busy) props.onClose();
+ }}
+ >
+
+
+
+ {pickMachine ? "Link a machine" : `Link ${targetName}`}
+
+
+ Agents here can then launch, message and wait on threads there. It appears in that
+ machine's Connections, where it can be revoked.
+
+ {!pickMachine && item.url !== undefined ? (
+
+ {item.url}
+
+ ) : null}
+
+ {pickMachine ? (
+
+ {machines.map((machine, index) => (
+ 0}
+ disabled={busy}
+ onPress={() => {
+ setChoice(machine.environmentId);
+ setError(null);
+ }}
+ />
+ ))}
+ 0}
+ disabled={busy}
+ onPress={() => {
+ setChoice(ANOTHER_ADDRESS);
+ setError(null);
+ }}
+ />
+
+ ) : null}
+ {needsAddress && !(item.url !== undefined && item.environmentId === undefined) ? (
+
+
+ {target.kind === "machine" ? (
+
+ This app has no address for {target.label} that another machine can reach. Enter its
+ https or Tailscale address.
+
+ ) : null}
+
+ ) : null}
+ {useExisting ? (
+
+ This environment is already linked to {targetName}. The agent can use that link as it
+ is; change its access in Settings → Connections.
+
+ ) : target.kind === "none" ? null : (
+ <>
+ {canMint ? null : (
+
+
+
+ Create a link on {targetName}: Settings → Connections → Create link, then paste
+ its code here.
+
+
+ )}
+
+ {ACCESS_CHOICES.map((choice, index) => (
+ 0}
+ disabled={busy}
+ onPress={() => setAccess(choice.access)}
+ />
+ ))}
+
+
+ An agent here never gets more there than its own mode here either. Threads it starts
+ there cannot change that environment's own threads, projects or settings.
+
+
+ )}
+ {error !== null ? (
+
+ {error}
+
+ ) : null}
+
+
+ );
+}
+
+/** One machine in the sheet's list: its glyph, name, and whether it is linked or reachable. */
+function MachineRow(props: {
+ readonly machine: LinkableMachine;
+ readonly selected: boolean;
+ readonly separated: boolean;
+ readonly disabled: boolean;
+ readonly onPress: () => void;
+}) {
+ const { machine } = props;
+ const status = machine.linked
+ ? "Linked"
+ : machine.connected
+ ? null
+ : machine.connectOnly
+ ? "T3 Connect"
+ : "Not connected";
+ return (
+
+
+
+
+ {machine.label}
+
+ {status === null ? null : {status}}
+
+ {props.selected ? (
+
+ ) : null}
+
+ );
+}
```
## apps/mobile/src/features/threads/ThreadFeed.tsx
Status: modified; +12; −0
```diff
@@ -12,6 +12,7 @@
} from "../../lib/enrichedLinkPresentation";
import { ThreadContextDivider } from "./thread-context-divider";
import { ThreadHandoffDivider, ThreadHandoffRow } from "./thread-handoff-row";
+import { LinkRequestCard } from "./LinkRequestCard";
import { SecretRequestCard } from "./SecretRequestCard";
import { AppActivityIndicator } from "../../components/ui/AppActivityIndicator";
import {
@@ -196,6 +197,7 @@
threadFeedRunIsUnsettled,
isContextCompactionActivityGroup,
isContextHandoffActivityGroup,
+ isLinkRequestActivityGroup,
isSecretRequestActivityGroup,
type ThreadFeedEntry,
type ThreadFeedLatestRun,
@@ -1285,6 +1287,16 @@
);
}
+ if (entry.type === "activity-group" && isLinkRequestActivityGroup(entry)) {
+ return (
+
+ );
+ }
+
if (entry.type === "activity-group" && isContextCompactionActivityGroup(entry)) {
const label = entry.activities[0]!.summary;
const active =
```
## apps/mobile/src/lib/threadActivity.ts
Status: modified; +13; −1
```diff
@@ -365,6 +365,12 @@
);
}
+export function isLinkRequestActivityGroup(entry: ThreadFeedActivityGroup): boolean {
+ return (
+ entry.activities.length === 1 && entry.activities[0]?.projectedItem.item.type === "link_request"
+ );
+}
+
function isUserInputActivityGroup(entry: ThreadFeedActivityGroup): boolean {
return entry.activities.some((activity) => activity.workEntry.questionAnswer !== undefined);
}
@@ -473,7 +479,8 @@
item.type === "thread_created" ||
item.type === "system_notice" ||
item.type === "proposed_plan" ||
- item.type === "secret_request"
+ item.type === "secret_request" ||
+ item.type === "link_request"
);
}
@@ -594,6 +601,8 @@
return "zap";
case "secret_request":
return "lock";
+ case "link_request":
+ return "globe";
}
}
@@ -650,6 +659,8 @@
return "Thread created";
case "secret_request":
return item.label;
+ case "link_request":
+ return `Link ${item.label ?? item.url ?? "an environment"}`;
case "dynamic_tool": {
const classified = classifyToolActivity({
itemType: "dynamic_tool_call",
@@ -708,6 +719,7 @@
case "thread_created":
return item.targetThreadId;
case "secret_request":
+ case "link_request":
return item.reason || null;
case "subagent":
return item.result ?? item.progress ?? item.prompt;
```
## apps/mobile/src/state/peerLinks.ts
Status: modified; +42; −0
```diff
@@ -1,5 +1,47 @@
+import { useAtomValue } from "@effect/atom-react";
+import { linkableMachines } from "@t3tools/client-runtime/link-request";
import { createPeerLinkEnvironmentAtoms } from "@t3tools/client-runtime/state/peerLinks";
+import type { EnvironmentId } from "@t3tools/contracts";
+import { useEffect, useEffectEvent, useMemo } from "react";
import { connectionAtomRuntime } from "../connection/runtime";
+import { useEnvironments } from "./environments";
+import { relayEnvironmentDiscovery } from "./relay";
+import { useEnvironmentQuery } from "./query";
+import { useAtomCommand } from "./use-atom-command";
export const peerLinkEnvironment = createPeerLinkEnvironmentAtoms(connectionAtomRuntime);
+
+/**
+ * The user's other machines, for picking one to link `environmentId` to,
+ * including those only on their T3 Connect account. Reads that environment's
+ * links only while `enabled`, since listing them asks every linked machine
+ * whether it answers.
+ */
+export function useLinkableMachines(environmentId: EnvironmentId, enabled: boolean) {
+ const { environments } = useEnvironments();
+ const discovery = useAtomValue(relayEnvironmentDiscovery.stateValueAtom);
+ const refreshDiscovery = useAtomCommand(relayEnvironmentDiscovery.refresh, {
+ reportFailure: false,
+ });
+ // Connect hosts are listed only once something asked for them, so ask on open.
+ const discoverIfUnasked = useEffectEvent(() => {
+ if (discovery.environments.size === 0 && !discovery.refreshing) void refreshDiscovery();
+ });
+ useEffect(() => {
+ if (enabled) discoverIfUnasked();
+ }, [enabled]);
+ const links = useEnvironmentQuery(
+ enabled ? peerLinkEnvironment.list({ environmentId, input: {} }) : null,
+ );
+ return useMemo(
+ () =>
+ linkableMachines({
+ environments,
+ threadEnvironmentId: environmentId,
+ linkedIds: new Set(links.data?.links.map((link) => link.environmentId)),
+ connect: discovery.environments,
+ }),
+ [discovery.environments, environmentId, environments, links.data],
+ );
+}
```
## apps/mobile/src/state/peerPairingCode.ts
Status: added; +24; −0
```diff
@@ -1,0 +1,24 @@
+import { createPeerPairingCodeCommand } from "@t3tools/client-runtime/state/peerPairingCode";
+import { squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime";
+import type { AuthMcpClientAccess, EnvironmentId } from "@t3tools/contracts";
+import * as Result from "effect/Result";
+
+import { connectionAtomRuntime } from "../connection/runtime";
+import { appAtomRegistry } from "./atom-registry";
+
+const mintPairingCode = createPeerPairingCodeCommand(connectionAtomRuntime);
+
+/**
+ * A fresh pairing code minted on `environmentId` with this client's session
+ * there, for linking with `access`. Fails with `PeerPairingCodeError` when the
+ * session may not create one.
+ */
+export async function mintPairingCodeOn(
+ environmentId: EnvironmentId,
+ access: AuthMcpClientAccess,
+): Promise> {
+ const result = await mintPairingCode.run(appAtomRegistry, { environmentId, input: access });
+ return result._tag === "Success"
+ ? Result.succeed(result.value)
+ : Result.fail(squashAtomCommandFailure(result));
+}
```
## apps/server/src/auth/EnvironmentAuth.ts
Status: modified; +1; −17
```diff
@@ -2,9 +2,6 @@
AuthAccessTokenType,
AuthAccessWriteScope,
AuthOrchestrationOperateScope,
- AuthOrchestrationReadScope,
- AuthSettingsWriteScope,
- AuthSourceControlWriteScope,
AuthStandardClientScopes,
type AuthAccessTokenResult,
type AuthBrowserSessionResult,
@@ -23,6 +20,7 @@
type AuthWebSocketTicketResult,
DpopFailureReason,
type DpopFailureReason as DpopFailureReasonType,
+ mcpClientScopes,
} from "@t3tools/contracts";
import { encodeOAuthScope } from "@t3tools/shared/oauthScope";
import * as Context from "effect/Context";
@@ -77,20 +75,6 @@
const MCP_CLIENT_SUBJECT = "mcp-client";
const MCP_CLIENT_SESSION_TTL = Duration.days(30);
-export const mcpClientScopes = (
- access: AuthMcpClientAccess,
-): ReadonlyArray =>
- access === "read-only"
- ? [AuthOrchestrationReadScope]
- : access === "full-access"
- ? [
- AuthOrchestrationReadScope,
- AuthOrchestrationOperateScope,
- AuthSettingsWriteScope,
- AuthSourceControlWriteScope,
- ]
- : [AuthOrchestrationReadScope, AuthOrchestrationOperateScope];
-
export interface McpClientSession {
readonly sessionId: AuthSessionId;
readonly label: string;
```
## apps/server/src/auth/McpOAuth.test.ts
Status: modified; +19; −0
```diff
@@ -1,8 +1,10 @@
import * as NodeServices from "@effect/platform-node/NodeServices";
import {
+ AuthMcpClientAccess,
AuthStandardClientScopes,
type AuthCreatePairingCredentialInput,
EnvironmentHttpApi,
+ mcpClientScopes,
} from "@t3tools/contracts";
import { expect, it } from "@effect/vitest";
import * as Context from "effect/Context";
@@ -480,6 +482,23 @@
),
);
+it.live("a code minted with an access level's scopes approves that access", () =>
+ withRoutes((handler, auth) =>
+ Effect.gen(function* () {
+ const params = authorizeParams(yield* registeredClientId(handler));
+ for (const access of AuthMcpClientAccess.literals) {
+ const code = yield* auth.issuePairingCredential({ scopes: [...mcpClientScopes(access)] });
+ const approval = yield* decide(handler, params, {
+ _tag: "pairing-code",
+ access,
+ code: code.credential,
+ });
+ expect(approval.status, access).toBe(200);
+ }
+ }),
+ ),
+);
+
it.live("one-click approves only access the browser session holds the scopes for", () =>
withRoutes((handler, auth) =>
Effect.gen(function* () {
```
## apps/server/src/auth/McpOAuth.ts
Status: modified; +3; −4
```diff
@@ -14,6 +14,7 @@
AuthOrchestrationOperateScope,
AuthOrchestrationReadScope,
MCP_PEER_LINK_SOFTWARE_ID,
+ mcpClientScopes,
} from "@t3tools/contracts";
import { encodeOAuthScope } from "@t3tools/shared/oauthScope";
import * as Clock from "effect/Clock";
@@ -446,9 +447,7 @@
if (
session === undefined ||
!timingSafeEqualBase64Url(decision.csrfToken, session.csrfToken) ||
- !EnvironmentAuth.mcpClientScopes(decision.access).every((scope) =>
- session.scopes.includes(scope),
+ !mcpClientScopes(decision.access).every((scope) => session.scopes.includes(scope))
- )
) {
return yield* new McpOAuthPageError({
description: "Your session cannot approve this request. Enter a pairing code instead.",
@@ -533,7 +532,7 @@
access_token: issued.token,
token_type: "Bearer",
expires_in: Math.max(0, Math.floor((issued.expiresAt.epochMilliseconds - now) / 1000)),
+ scope: encodeOAuthScope(mcpClientScopes(pending.access)),
- scope: encodeOAuthScope(EnvironmentAuth.mcpClientScopes(pending.access)),
};
},
);
```
## apps/server/src/auth/mcpOAuthHttp.ts
Status: modified; +2; −4
```diff
@@ -3,6 +3,7 @@
type AuthMcpAuthorizationRequest,
AuthMcpClientAccess,
EnvironmentHttpApi,
+ mcpClientScopes,
} from "@t3tools/contracts";
import * as Effect from "effect/Effect";
import * as Result from "effect/Result";
@@ -10,7 +11,6 @@
import * as HttpEffect from "effect/http/HttpEffect";
import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
-import * as EnvironmentAuth from "./EnvironmentAuth.ts";
import * as McpOAuth from "./McpOAuth.ts";
import { renderErrorPage } from "./mcpOAuthHtml.ts";
@@ -121,9 +121,7 @@
: {
csrfToken: session.csrfToken,
oneClickAccess: AuthMcpClientAccess.literals.filter((access) =>
- EnvironmentAuth.mcpClientScopes(access).every((scope) =>
- session.scopes.includes(scope),
+ mcpClientScopes(access).every((scope) => session.scopes.includes(scope)),
- ),
),
}),
};
```
## apps/server/src/mcp/OrchestratorMcpService.ts
Status: modified; +4; −0
```diff
@@ -901,6 +901,10 @@
return `Created thread ${item.targetThreadId} with ${item.targetProviderInstanceId} (${item.targetModel}).`;
case "secret_request":
return `Asked the user for ${item.label}: ${item.secretStatus}.`;
+ case "link_request":
+ return item.linkStatus === "linked"
+ ? `Linked ${item.linkedLabel ?? item.label ?? item.url ?? "an environment"} (${item.linkedAccess}).`
+ : `Asked the user to link ${item.label ?? item.url ?? "an environment"}: ${item.linkStatus}.`;
case "subagent":
return item.result ?? item.progress ?? item.prompt;
case "dynamic_tool":
```
## apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts
Status: modified; +4; −0
```diff
@@ -48,6 +48,8 @@
import * as McpHttpServer from "./McpHttpServer.ts";
import * as McpInvocationContext from "./McpInvocationContext.ts";
import * as PeerForwarding from "../peer/PeerForwarding.ts";
+import * as PeerLinkRequests from "../peer/PeerLinkRequests.ts";
+import * as PeerLinks from "../peer/PeerLinks.ts";
const layerOrchestratorToolkit = McpHttpServer.layerOrchestratorToolkit.pipe(
Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})),
@@ -229,6 +231,8 @@
Layer.provide(ServerSettings.layerTest()),
Layer.provide(layerUnusedScheduledTaskStub),
Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
+ Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})),
+ Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})),
Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
Layer.provideMerge(
SecretRequests.layer.pipe(
```
## apps/server/src/mcp/linkOrigin.test.ts
Status: modified; +4; −0
```diff
@@ -44,6 +44,8 @@
import * as McpInvocationContext from "./McpInvocationContext.ts";
import * as PeerForwarding from "../peer/PeerForwarding.ts";
import * as ThreadSearch from "../orchestration-v2/ThreadSearch.ts";
+import * as PeerLinkRequests from "../peer/PeerLinkRequests.ts";
+import * as PeerLinks from "../peer/PeerLinks.ts";
// A linked environment's session drives this one's real orchestrator through
// its real T3 tools. What the link starts carries its origin, and the link
@@ -156,6 +158,8 @@
),
Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
+ Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})),
+ Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})),
Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})),
Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})),
Layer.provide(
```
## apps/server/src/mcp/toolkits/core.test.ts
Status: modified; +12; −0
```diff
@@ -53,6 +53,8 @@
import * as EnvironmentHandlers from "./environment/handlers.ts";
import * as AttachmentHandlers from "./attachment/handlers.ts";
import * as PeerForwarding from "../../peer/PeerForwarding.ts";
+import * as PeerLinkRequests from "../../peer/PeerLinkRequests.ts";
+import * as PeerLinks from "../../peer/PeerLinks.ts";
const layerOrchestratorToolkit = McpHttpServer.layerOrchestratorToolkit.pipe(
Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})),
@@ -74,6 +76,8 @@
Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})),
Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
+ Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})),
+ Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})),
);
const threadId = ThreadId.make("mcp-core-thread");
@@ -546,6 +550,8 @@
Layer.provide(Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({})),
Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
+ Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})),
+ Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})),
Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})),
@@ -603,6 +609,8 @@
Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
+ Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})),
+ Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})),
Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})),
),
),
@@ -690,6 +698,8 @@
Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})),
Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
+ Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})),
+ Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})),
),
),
),
@@ -781,6 +791,8 @@
Layer.provide(Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({})),
Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
+ Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})),
+ Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})),
Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})),
```
## apps/server/src/mcp/toolkits/orchestrator/handlers.ts
Status: modified; +25; −0
```diff
@@ -3,6 +3,8 @@
import * as Effect from "effect/Effect";
import * as PeerForwarding from "../../../peer/PeerForwarding.ts";
+import * as PeerLinkRequests from "../../../peer/PeerLinkRequests.ts";
+import * as PeerLinks from "../../../peer/PeerLinks.ts";
import { readCaller } from "../../threadAccess.ts";
import * as McpInvocationContext from "../../McpInvocationContext.ts";
import * as McpToolAccess from "../../McpToolAccess.ts";
@@ -150,6 +152,29 @@
return yield* forwarding.links(scope);
}),
),
+ // A link lets every agent here reach the other environment, so it is an
+ // environment setting: full access, and never for linked work.
+ t3_environment_link: McpToolAccess.writesEnvironment((input) =>
+ Effect.gen(function* () {
+ const scope = yield* McpInvocationContext.McpInvocationContext;
+ const requests = yield* PeerLinkRequests.PeerLinkRequests;
+ return yield* requests.request(scope, input);
+ }),
+ ),
+ t3_environment_unlink: McpToolAccess.writesEnvironment(({ environmentId }) =>
+ Effect.gen(function* () {
+ const links = yield* PeerLinks.PeerLinks;
+ const removed = yield* links
+ .unlink(environmentId)
+ .pipe(
+ Effect.mapError(
+ (error) =>
+ new OrchestratorMcpFailure({ code: "orchestration_error", message: error.message }),
+ ),
+ );
+ return { environmentId, removed };
+ }),
+ ),
t3_thread_interrupt: McpToolAccess.writesThreads(
(input) => [input.threadId],
(input) =>
```
## apps/server/src/mcp/toolkits/orchestrator/tools.ts
Status: modified; +41; −1
```diff
@@ -2,7 +2,11 @@
OrchestratorMcpCapabilitiesInput,
OrchestratorMcpCapabilitiesResult,
OrchestrationV2StopWorkflowInput,
+ OrchestratorMcpEnvironmentLinkInput,
+ OrchestratorMcpEnvironmentLinkResult,
OrchestratorMcpEnvironmentLinksResult,
+ OrchestratorMcpEnvironmentUnlinkInput,
+ OrchestratorMcpEnvironmentUnlinkResult,
OrchestratorMcpCreateThreadsInput,
OrchestratorMcpCreateThreadsResult,
OrchestratorMcpDelegateTaskInput,
@@ -40,6 +44,8 @@
import * as ThreadManagementService from "../../../orchestration-v2/ThreadManagementService.ts";
import * as PeerForwarding from "../../../peer/PeerForwarding.ts";
+import * as PeerLinkRequests from "../../../peer/PeerLinkRequests.ts";
+import * as PeerLinks from "../../../peer/PeerLinks.ts";
import * as McpInvocationContext from "../../McpInvocationContext.ts";
import * as OrchestratorMcpService from "../../OrchestratorMcpService.ts";
import * as ThreadMetadataMcpService from "../../ThreadMetadataMcpService.ts";
@@ -51,6 +57,12 @@
];
/** Listing this environment's links. */
const linkDependencies = [...dependencies, PeerForwarding.PeerForwarding];
+/** Making and forgetting links. */
+const linkingDependencies = [
+ ...dependencies,
+ PeerLinkRequests.PeerLinkRequests,
+ PeerLinks.PeerLinks,
+];
const threadMetadataDependencies = [
McpInvocationContext.McpInvocationContext,
ThreadManagementService.ThreadManagementService,
@@ -288,7 +300,7 @@
const EnvironmentLinksTool = Tool.make("t3_environment_links", {
description:
+ "List the other T3 Code environments this one is linked to, such as the user's other machines, and whether each answers now. A machine the user names that is not listed is not linked yet: call t3_environment_link for it rather than looking elsewhere. Pass one's environmentId to any tool that takes environmentId, such as t3_thread_launch, t3_thread_send, t3_pending_request_respond, t3_queue_list or t3_project_list, to act there; delegate_task takes it as target.environmentId. A linked environment lets this one's agents do at most its access there, and never more than the calling agent may do here. An expired link has to be linked again from a new pairing code.",
- "List the other T3 Code environments this one is linked to, such as the user's other machines, and whether each answers now. Pass one's environmentId to any tool that takes environmentId, such as t3_thread_launch, t3_thread_send, t3_pending_request_respond, t3_queue_list or t3_project_list, to act there; delegate_task takes it as target.environmentId. A linked environment lets this one's agents do at most its access there, and never more than the calling agent may do here. An expired link has to be linked again from a new pairing code.",
success: OrchestratorMcpEnvironmentLinksResult,
failure: OrchestratorMcpFailure,
failureMode: "return",
@@ -299,6 +311,32 @@
.annotate(Tool.Destructive, false)
.annotate(Tool.Idempotent, true);
+const EnvironmentLinkTool = Tool.make("t3_environment_link", {
+ description:
+ 'Ask the user to link another T3 Code environment, such as their VPS or another machine, so agents here can work there (t3_environment_links). Call it when the user asks to link, connect, or delegate to another machine. You usually do not know where it answers, so pass no address: pass a hint from their words ("vps") and the card lets the user pick the machine. Pass environmentId when a machine the user mentioned names it; if it is already linked, this returns linked at once. Pass url only when the user gave an address. The card lets the user pick the access agents here get there and approve it; their app signs in there with their own session, or they paste a pairing code from there into the card. It returns pending at once: the user answers in the card whenever they get to it, and the outcome (linked with the environmentId to pass to other tools, declined, or failed with the reason) arrives as a message in this thread. End your turn or keep working; do not poll or call it again to check. Never ask for a pairing code in chat.',
+ parameters: OrchestratorMcpEnvironmentLinkInput,
+ success: OrchestratorMcpEnvironmentLinkResult,
+ failure: OrchestratorMcpFailure,
+ failureMode: "return",
+ dependencies: linkingDependencies,
+})
+ .annotate(Tool.Title, "Link another environment")
+ .annotate(Tool.Destructive, false)
+ .annotate(Tool.OpenWorld, true);
+
+const EnvironmentUnlinkTool = Tool.make("t3_environment_unlink", {
+ description:
+ "Forget a linked environment, so agents here stop working there. The session there stays listed in that environment's Connections until it is revoked there.",
+ parameters: OrchestratorMcpEnvironmentUnlinkInput,
+ success: OrchestratorMcpEnvironmentUnlinkResult,
+ failure: OrchestratorMcpFailure,
+ failureMode: "return",
+ dependencies: linkingDependencies,
+})
+ .annotate(Tool.Title, "Forget a linked environment")
+ .annotate(Tool.Destructive, true)
+ .annotate(Tool.Idempotent, true);
+
export const OrchestratorToolkit = Toolkit.make(
OrchestratorCapabilitiesTool,
DelegateTaskTool,
@@ -318,4 +356,6 @@
ThreadWaitTool,
ThreadInterruptTool,
EnvironmentLinksTool,
+ EnvironmentLinkTool,
+ EnvironmentUnlinkTool,
);
```
## apps/server/src/observability/RpcInstrumentation.ts
Status: modified; +1; −0
```diff
@@ -62,6 +62,7 @@
[WS_METHODS.peerLinksList]: "peerLinks",
[WS_METHODS.peerLinksLink]: "peerLinks",
[WS_METHODS.peerLinksUnlink]: "peerLinks",
+ [WS_METHODS.peerLinksAnswerRequest]: "peerLinks",
[WS_METHODS.serverUpdateSettings]: "server",
[WS_METHODS.serverSearchAcpRegistry]: "server",
[WS_METHODS.serverPrepareAcpRegistryAgent]: "server",
```
## apps/server/src/orchestration-v2/Orchestrator.ts
Status: modified; +201; −0
```diff
@@ -500,6 +500,7 @@
case "thread.created.record":
return command.parentThreadId;
case "secret_request.record":
+ case "link_request.record":
return command.threadId;
case "thread.fork":
case "thread.merge_back":
@@ -3665,6 +3666,7 @@
now,
cancelQueuedDelivery: false,
});
+ yield* closePendingLinkRequests({ command, events, threadId: command.threadId, now });
}
// Settle joins archive here: both mean "done with this
@@ -8008,6 +8010,201 @@
},
);
+ /**
+ * Records or updates the card for a link an agent asked the user to make.
+ * The pairing code goes straight to the link and never through orchestration.
+ */
+ const dispatchLinkRequestRecord = Effect.fn("orchestrationV2.dispatch.linkRequestRecord")(
+ function* (
+ command: Extract,
+ events: Ref.Ref>,
+ effects: Ref.Ref>,
+ ) {
+ const projection = yield* projectionStore
+ .getThreadRecords(
+ command.threadId,
+ ["runs", "nodes", "turnItems", "attempts", "providerTurns"],
+ { turnItemTypes: ["link_request"], messageRoles: [] },
+ )
+ .pipe(
+ Effect.mapError(
+ (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }),
+ ),
+ );
+ const run = projection.runs.find((candidate) => candidate.id === command.runId);
+ if (run === undefined) {
+ return yield* new OrchestratorDispatchError({
+ commandId: command.commandId,
+ commandType: command.type,
+ cause: `Run ${command.runId} was not found.`,
+ });
+ }
+ const existing = projection.turnItems.find((item) => item.id === command.turnItemId);
+ if (existing !== undefined && existing.type !== "link_request") {
+ return yield* new OrchestratorDispatchError({
+ commandId: command.commandId,
+ commandType: command.type,
+ cause: `Turn item ${command.turnItemId} is not a link request.`,
+ });
+ }
+ if (existing !== undefined && existing.runId !== command.runId) {
+ return yield* new OrchestratorDispatchError({
+ commandId: command.commandId,
+ commandType: command.type,
+ cause: `Link request ${command.turnItemId} belongs to another run.`,
+ });
+ }
+ // Only a new card needs its run's current root and a live run. An answer
+ // may land long after a steer restarted the run or the run ended, and the
+ // link it reports is already made.
+ if (existing === undefined && run.rootNodeId !== command.nodeId) {
+ return yield* new OrchestratorDispatchError({
+ commandId: command.commandId,
+ commandType: command.type,
+ cause: `Node ${command.nodeId} is not the root of run ${command.runId}.`,
+ });
+ }
+ if (
+ existing === undefined &&
+ (!isBlockingRun(run) || (yield* stopReachedRun(command, command.threadId, run.id)))
+ ) {
+ return yield* new OrchestratorDispatchError({
+ commandId: command.commandId,
+ commandType: command.type,
+ cause: `Run ${command.runId} is not active.`,
+ });
+ }
+ const nodeId = existing?.nodeId ?? command.nodeId;
+ const now = yield* DateTime.now;
+ // A request is answered once, and a retry finds its card as it was
+ // asked: either one records the card unchanged.
+ const unchanged =
+ existing !== undefined &&
+ (existing.linkStatus !== "pending" || command.linkStatus === "pending");
+ const pending = command.linkStatus === "pending";
+ const {
+ type: _type,
+ commandId: _commandId,
+ turnItemId: _turnItemId,
+ wake,
+ ...fields
+ } = command;
+ const turnItem: OrchestrationV2TurnItem = unchanged
+ ? existing
+ : {
+ ...fields,
+ id: command.turnItemId,
+ // An update keeps where the card was asked, even after a restart.
+ ...(existing === undefined
+ ? {
+ providerThreadId: run.providerThreadId,
+ providerTurnId: providerTurnForRun(projection, run)?.id ?? null,
+ }
+ : {
+ nodeId: existing.nodeId,
+ providerThreadId: existing.providerThreadId,
+ providerTurnId: existing.providerTurnId,
+ }),
+ nativeItemRef: null,
+ parentItemId: null,
+ ordinal: existing?.ordinal ?? (yield* nextTurnItemOrdinal(projection)),
+ status: pending
+ ? "waiting"
+ : command.linkStatus === "linked"
+ ? "completed"
+ : command.linkStatus === "failed"
+ ? "failed"
+ : "cancelled",
+ title: command.label ?? command.url ?? "Link an environment",
+ startedAt: existing?.startedAt ?? now,
+ completedAt: pending ? null : now,
+ updatedAt: now,
+ type: "link_request",
+ };
+ yield* emit(
+ events,
+ command,
+ )({
+ type: "turn-item.updated",
+ threadId: command.threadId,
+ runId: command.runId,
+ nodeId,
+ providerInstanceId: run.providerInstanceId,
+ occurredAt: now,
+ payload: turnItem,
+ });
+ // The agent asked without waiting, so the answer reaches it as a
+ // notification: it starts a turn, or follows the one running. Only the
+ // answer that closes the card wakes it, and never a closed thread.
+ if (
+ wake === undefined ||
+ unchanged ||
+ projection.thread.archivedAt !== null ||
+ projection.thread.deletedAt !== null
+ ) {
+ return;
+ }
+ yield* dispatchMessage(
+ {
+ type: "message.dispatch",
+ commandId: command.commandId,
+ threadId: command.threadId,
+ messageId: wake.messageId,
+ text: wake.text,
+ notification: wake.notification,
+ attachments: [],
+ dispatchMode: { type: "queue_after_active" },
+ createdBy: "agent",
+ creationSource: "server",
+ },
+ events,
+ effects,
+ );
+ },
+ );
+
+ /**
+ * Closes the link cards still waiting for the user. Their agent stopped
+ * waiting long ago, so nobody is woken: stopping or archiving the thread
+ * means the user is done with it.
+ */
+ const closePendingLinkRequests = Effect.fn("orchestrationV2.closePendingLinkRequests")(
+ function* (input: {
+ readonly command: OrchestrationV2ServerCommand;
+ readonly events: Ref.Ref>;
+ readonly threadId: ThreadId;
+ readonly now: DateTime.Utc;
+ }) {
+ const { turnItems } = yield* projectionStore
+ .getThreadRecords(input.threadId, ["turnItems"], {
+ turnItemTypes: ["link_request"],
+ turnItemStatuses: ["waiting"],
+ messageRoles: [],
+ })
+ .pipe(mapDispatchError(input.command));
+ for (const item of turnItems) {
+ if (item.type !== "link_request" || item.linkStatus !== "pending") continue;
+ yield* emit(
+ input.events,
+ input.command,
+ )({
+ type: "turn-item.updated",
+ threadId: input.threadId,
+ ...(item.runId === null ? {} : { runId: item.runId }),
+ ...(item.nodeId === null ? {} : { nodeId: item.nodeId }),
+ occurredAt: input.now,
+ payload: {
+ ...item,
+ status: "cancelled",
+ linkStatus: "cancelled",
+ completedAt: input.now,
+ updatedAt: input.now,
+ },
+ });
+ }
+ },
+ );
+
const dispatchRuntimeRequestRespond = Effect.fn("Orchestrator.dispatchRuntimeRequestRespond")(
function* (
command: Extract,
@@ -9643,6 +9840,7 @@
now: input.now,
});
}
+ yield* closePendingLinkRequests({ ...input, threadId: thread.id });
});
const dispatchBackgroundWorkSettle = Effect.fn("Orchestrator.dispatchBackgroundWorkSettle")(
@@ -11942,6 +12140,9 @@
case "secret_request.record":
yield* dispatchSecretRequestRecord(command, events);
break;
+ case "link_request.record":
+ yield* dispatchLinkRequestRecord(command, events, effects);
+ break;
default:
return yield* dispatchUnsupported(command);
}
```
## apps/server/src/orchestration-v2/ProjectionStore.ts
Status: modified; +40; −10
```diff
@@ -1083,6 +1083,7 @@
readonly blocking_failure_payload_json: string | null;
readonly pending_request_payload_json: string | null;
readonly pending_secret_request_payload_json: string | null;
+ readonly pending_link_request_payload_json: string | null;
readonly latest_user_message_at: string | null;
readonly latest_user_authored_message_at: string | null;
readonly has_actionable_proposed_plan: number;
@@ -1504,7 +1505,8 @@
function secretRequestAsPendingInput(
item: OrchestrationV2TurnItem | null,
): OrchestrationV2PendingRuntimeRequestSummary | null {
- if (item?.type !== "secret_request" || item.nodeId === null) return null;
+ if ((item?.type !== "secret_request" && item?.type !== "link_request") || item.nodeId === null)
+ return null;
return {
id: RuntimeRequestId.make(item.id),
kind: "user_input",
@@ -1513,6 +1515,14 @@
};
}
+/**
+ * A secret card is answered while its agent waits in the run. A link card's
+ * agent asked without waiting, so the card waits on the user after its run.
+ */
+const waitsOnUser = (item: OrchestrationV2TurnItem, liveRunIds: ReadonlySet) =>
+ item.type === "link_request" ||
+ (item.type === "secret_request" && item.runId !== null && liveRunIds.has(item.runId));
+
export function threadShellFromProjection(
projection: OrchestrationV2ThreadProjection,
): OrchestrationV2ThreadShell {
@@ -1550,13 +1560,7 @@
pendingRuntimeRequest =
secretRequestAsPendingInput(
projection.turnItems
+ .filter((item) => item.status === "waiting" && waitsOnUser(item, liveRunIds))
- .filter(
- (item) =>
- item.type === "secret_request" &&
- item.status === "waiting" &&
- item.runId !== null &&
- liveRunIds.has(item.runId),
- )
.toSorted(
(left, right) =>
DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt),
@@ -5763,6 +5767,15 @@
LIMIT 1
) AS pending_secret_request_payload_json,
(
+ SELECT link.payload_json
+ FROM orchestration_v2_projection_turn_items link
+ INDEXED BY orchestration_v2_projection_turn_items_waiting_link_idx
+ WHERE link.thread_id = t.thread_id
+ AND link.type = 'link_request' AND link.status = 'waiting'
+ ORDER BY link.updated_at DESC, link.turn_item_id DESC
+ LIMIT 1
+ ) AS pending_link_request_payload_json,
+ (
SELECT message.updated_at
FROM orchestration_v2_projection_messages message
WHERE message.thread_id = t.thread_id
@@ -6038,6 +6051,12 @@
SELECT 1 FROM orchestration_v2_projection_runtime_requests request
WHERE request.thread_id = t.thread_id AND request.status = 'pending'
)
+ AND NOT EXISTS (
+ SELECT 1 FROM orchestration_v2_projection_turn_items link
+ INDEXED BY orchestration_v2_projection_turn_items_waiting_link_idx
+ WHERE link.thread_id = t.thread_id
+ AND link.type = 'link_request' AND link.status = 'waiting'
+ )
ORDER BY t.updated_at ASC, t.thread_id ASC
`;
if (rows.length === 0) return [];
@@ -6196,6 +6215,14 @@
pendingTurnItemsByThreadId,
} = input;
const thread = yield* decodeThreadPayload(row.payload_json);
+ const waitingCards = [
+ row.pending_secret_request_payload_json === null
+ ? null
+ : yield* decodeTurnItemPayload(row.pending_secret_request_payload_json),
+ row.pending_link_request_payload_json === null
+ ? null
+ : yield* decodeTurnItemPayload(row.pending_link_request_payload_json),
+ ].filter((item) => item !== null);
let pendingRuntimeRequest =
row.pending_request_payload_json !== null
? pendingRuntimeRequestSummary(
@@ -6204,11 +6231,14 @@
: null;
if (
(pendingRuntimeRequest === null || pendingRuntimeRequest.blocking === false) &&
- row.pending_secret_request_payload_json !== null
+ waitingCards.length > 0
) {
pendingRuntimeRequest =
secretRequestAsPendingInput(
+ waitingCards.toSorted(
- yield* decodeTurnItemPayload(row.pending_secret_request_payload_json),
+ (left, right) =>
+ DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt),
+ )[0] ?? null,
) ?? pendingRuntimeRequest;
}
let terminalFailureItem =
```
## apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts
Status: modified; +83; −0
```diff
@@ -418,6 +418,89 @@
}).pipe(Effect.provide(layer));
});
+it.effect("recovery closes a secret card's form but leaves a link card open", () => {
+ const threadId = ThreadId.make("thread_secret_recovery");
+ const runId = RunId.make("run_secret_recovery");
+ let committedInput: Parameters[0] | null =
+ null;
+ const projection = {
+ thread: { id: threadId },
+ runtimeRequests: [],
+ providerSessions: [],
+ providerThreads: [],
+ providerTurns: [],
+ runs: [{ id: runId, status: "running", providerInstanceId: ProviderInstanceId.make("codex") }],
+ attempts: [],
+ nodes: [],
+ subagents: [],
+ messages: [],
+ turnItems: [
+ {
+ id: "turn-item:secret-request:recovery",
+ threadId,
+ runId,
+ nodeId: null,
+ type: "secret_request",
+ status: "waiting",
+ secretStatus: "pending",
+ label: "GitHub token",
+ reason: "Used as GH_TOKEN.",
+ },
+ {
+ id: "turn-item:link-request:recovery",
+ threadId,
+ runId,
+ nodeId: null,
+ type: "link_request",
+ status: "waiting",
+ linkStatus: "pending",
+ url: "https://box.example.ts.net",
+ reason: "Run the suite there.",
+ },
+ ],
+ } as unknown as OrchestrationV2ThreadProjection;
+ const layer = ProviderRuntimeRecovery.layer.pipe(
+ Layer.provide(ServerSettings.layerTest()),
+ Layer.provide(
+ Layer.mergeAll(
+ Layer.mock(ProjectionStore.ProjectionStoreV2)({
+ getRecoveryThreadIds: () => Effect.succeed([threadId]),
+ getRuntimeRecoveryProjection: () => Effect.succeed(projection),
+ }),
+ Layer.mock(EventSink.EventSinkV2)({
+ commitCommand: (input) => {
+ committedInput = input;
+ return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);
+ },
+ }),
+ IdAllocator.layer,
+ Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
+ runRecoveryOnce: Effect.succeed(false),
+ }),
+ Layer.mock(EffectOutbox.EffectOutboxV2)({
+ listByCommandId: () => Effect.succeed([]),
+ reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
+ }),
+ ),
+ ),
+ );
+
+ return Effect.gen(function* () {
+ yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");
+ const cards = (committedInput?.events ?? []).flatMap((event) =>
+ event.type === "turn-item.updated" ? [event.payload] : [],
+ );
+ const secret = cards.find((card) => card.type === "secret_request");
+ assert.equal(secret?.status, "cancelled");
+ assert.equal(secret?.type === "secret_request" ? secret.secretStatus : null, "cancelled");
+ // Nothing in memory waits on a link card; the user can still answer it.
+ assert.equal(
+ cards.some((card) => card.type === "link_request"),
+ false,
+ );
+ }).pipe(Effect.provide(layer));
+});
+
it.effect.each(["queued", "preparing"] as const)("recovers $0 work after restart", (status) => {
const threadId = ThreadId.make("thread_queued_restart");
const runId = RunId.make("run_queued_restart");
```
## apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts
Status: modified; +2; −0
```diff
@@ -444,6 +444,8 @@
candidate.runId === run.id &&
(candidate.nodeId === null || !messageRequestNodeIds.has(candidate.nodeId)) &&
!isAppOwnedDelegationItem(candidate) &&
+ // A link card waits on the user, not the run, and answers by message.
+ candidate.type !== "link_request" &&
(candidate.status === "pending" ||
candidate.status === "running" ||
candidate.status === "waiting"),
```
[Next page](https://git.cafe/pingdotgg/t3code/pull/17129/diffs?after=ImFwcHMvc2VydmVyL3NyYy9vcmNoZXN0cmF0aW9uLXYyL1Byb3ZpZGVyUnVudGltZVJlY292ZXJ5U2VydmljZS50cyI&expectedVersion=32&format=markdown)