Skip to content
Closed
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
28 changes: 22 additions & 6 deletions apps/mobile/src/features/threads/ThreadFeed.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -1333,6 +1333,8 @@ function useMarkdownStyles(): MarkdownStyleSets {
function AgentMessageFoldHeader(props: {
readonly environmentId: EnvironmentId;
readonly senderThreadId?: ThreadId;
/** A linked environment's sender title, for when this client cannot reach it. */
readonly senderTitle?: string;
readonly summary: string;
readonly expanded: boolean;
readonly onToggle: () => void;
Expand All @@ -1341,7 +1343,7 @@ function AgentMessageFoldHeader(props: {
const senderShell = useQuietThreadShell(
props.senderThreadId ? scopeThreadRef(props.environmentId, props.senderThreadId) : null,
);
const label = senderShell?.title.trim() || "Agent";
const label = senderShell?.title.trim() || props.senderTitle?.trim() || "Agent";
return (
<Pressable
accessibilityRole="button"
Expand Down Expand Up @@ -1384,8 +1386,14 @@ function AgentMessageFoldHeader(props: {
function OpenSenderThreadButton(props: {
readonly environmentId: EnvironmentId;
readonly senderThreadId: ThreadId;
/** In a linked environment, shown only once this client can reach the thread. */
readonly linked?: boolean;
}) {
const openLinkedThread = useOpenLinkedThread();
const linkedShell = useQuietThreadShell(
props.linked ? scopeThreadRef(props.environmentId, props.senderThreadId) : null,
);
if (props.linked && linkedShell === null) return null;
return (
<Pressable
accessibilityRole="link"
Expand Down Expand Up @@ -1916,6 +1924,12 @@ function renderFeedEntry(
if (isUser) {
const enterAnimated = isFreshTimestamp(message.createdAt);
const foldRowId = presentation.attribution === "agent" ? `agent:${message.id}` : null;
// A linked environment's agent sent it from a thread there.
const senderRef =
message.linkedSender ??
(message.senderThreadId
? scopeThreadRef(props.environmentId, message.senderThreadId)
: null);
const foldExpanded = foldRowId !== null && (props.expandedWorkRows[foldRowId] ?? false);
const bubbleBodyVisible = foldRowId === null || foldExpanded;
const intentBadge = props.userMessageGrouping.continuesSteer
Expand Down Expand Up @@ -2036,8 +2050,9 @@ function renderFeedEntry(
{presentation.attribution === "guest" ? <GuestBubbleTint /> : null}
{foldRowId !== null ? (
<AgentMessageFoldHeader
environmentId={props.environmentId}
senderThreadId={message.senderThreadId}
environmentId={senderRef?.environmentId ?? props.environmentId}
senderThreadId={senderRef?.threadId}
{...(message.linkedSender ? { senderTitle: message.linkedSender.title } : {})}
summary={
replaceComposerContextReferences(presentation.text, (ref) => ref.label)
.split(/\r?\n/)
Expand Down Expand Up @@ -2144,10 +2159,11 @@ function renderFeedEntry(
) : null}
{bubbleBodyVisible ? (
<View className="mt-1 flex-row items-center justify-end gap-1 pr-0.5">
{foldRowId !== null && message.senderThreadId ? (
{foldRowId !== null && senderRef ? (
<OpenSenderThreadButton
environmentId={props.environmentId}
senderThreadId={message.senderThreadId}
environmentId={senderRef.environmentId}
senderThreadId={senderRef.threadId}
linked={message.linkedSender !== undefined}
/>
) : null}
{intentBadge ? (
Expand Down
6 changes: 4 additions & 2 deletions apps/mobile/src/features/threads/ThreadQueueControl.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -70,8 +70,10 @@ function QueuedRunSender(props: {
}) {
const { attribution, scheduledTaskId } = resolveUserMessagePresentation(props.message);
const senderThreadId = attribution === "agent" ? props.message.senderThreadId : undefined;
// A linked environment's agent sent it from a thread there.
const linkedSender = attribution === "agent" ? props.message.linkedSender : undefined;
const senderShell = useQuietThreadShell(
senderThreadId ? scopeThreadRef(props.environmentId, senderThreadId) : null,
linkedSender ?? (senderThreadId ? scopeThreadRef(props.environmentId, senderThreadId) : null),
);
const tasks = useEnvironmentQuery(
scheduledTaskId
Expand All @@ -88,7 +90,7 @@ function QueuedRunSender(props: {
: attribution === "agent"
? {
icon: "bubble.left.and.bubble.right" as const,
label: senderShell?.title.trim() || "Agent",
label: senderShell?.title.trim() || linkedSender?.title.trim() || "Agent",
}
: attribution === "t3code"
? { icon: null, label: "Sent by T3 Code" }
Expand Down
2 changes: 2 additions & 0 deletions apps/mobile/src/lib/threadActivity.ts
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,7 @@ export interface ThreadFeedMessage {
readonly creationSource?: OrchestrationV2CreationSource;
readonly scheduledTaskId?: ScheduledTaskId;
readonly senderThreadId?: ThreadId;
readonly linkedSender?: import("@t3tools/contracts").OrchestrationV2LinkedSender;
readonly idleCompaction?: import("@t3tools/contracts").OrchestrationV2IdleCompaction;
readonly visibility: OrchestrationV2ProjectedTurnItem["visibility"];
readonly sourceThreadId: ThreadId;
Expand Down Expand Up @@ -2387,6 +2388,7 @@ export function buildThreadFeed(
creationSource: item.creationSource,
...(item.scheduledTaskId ? { scheduledTaskId: item.scheduledTaskId } : {}),
...(item.senderThreadId ? { senderThreadId: item.senderThreadId } : {}),
...(item.linkedSender ? { linkedSender: item.linkedSender } : {}),
...(item.idleCompaction ? { idleCompaction: item.idleCompaction } : {}),
}
: {}),
Expand Down
31 changes: 30 additions & 1 deletion apps/server/src/mcp/McpHttpServer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import {
ProviderInteractionMode,
type RunId,
RuntimeMode,
OrchestrationV2LinkedSender,
ThreadId,
} from "@t3tools/contracts";

Expand Down Expand Up @@ -190,6 +191,34 @@ const withModeLimit = (
: { ...invocation, client: { ...invocation.client, narrowedTo: limit.value } };
};

const decodeSenderThread = Schema.decodeUnknownOption(OrchestrationV2LinkedSender);

/** Records the thread a linked environment calls for; a malformed header only loses the name. */
const withSenderThread = (
invocation: McpInvocationContext.McpInvocationScope,
request: HttpServerRequest.HttpServerRequest,
): McpInvocationContext.McpInvocationScope => {
const header = request.headers[McpInvocationContext.SENDER_THREAD_HEADER];
if (header === undefined || invocation.client?.linked !== true) return invocation;
const [environmentId, threadId, title, ...rest] = header.trim().split("/");
if (environmentId === undefined || threadId === undefined || title === undefined)
return invocation;
if (rest.length > 0) return invocation;
let decoded: Option.Option<OrchestrationV2LinkedSender>;
try {
decoded = decodeSenderThread({
environmentId: decodeURIComponent(environmentId),
threadId: decodeURIComponent(threadId),
title: decodeURIComponent(title),
});
} catch {
return invocation;
}
return Option.isNone(decoded)
? invocation
: { ...invocation, client: { ...invocation.client, senderThread: decoded.value } };
};

// The Linked environments beta is off here: links wait until it is on again.
const linkedEnvironmentsOff = HttpServerResponse.jsonUnsafe(
{
Expand Down Expand Up @@ -235,7 +264,7 @@ const makeMcpAuthMiddleware = Effect.gen(function* () {
offerOAuth: Option.isSome(clients) && !looksLikeProviderToken(token),
});
}
const asked = withModeLimit(invocation, request);
const asked = withModeLimit(withSenderThread(invocation, request), request);
if (asked === undefined) return invalidModeLimit;
const limited = McpInvocationContext.withRuntimeModeCeiling(asked, runtimeModeCeiling);
const linkedEnvironments = yield* linksEnabled;
Expand Down
10 changes: 10 additions & 0 deletions apps/server/src/mcp/McpInvocationContext.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import {
PreviewAutomationUnavailableError,
type ProviderInstanceId,
type ProviderInteractionMode,
type OrchestrationV2LinkedSender,
type RuntimeMode,
type ThreadId,
} from "@t3tools/contracts";
Expand Down Expand Up @@ -45,6 +46,8 @@ export interface McpClientCaller {
* another environment passing on the limits of the agent it calls for.
*/
readonly narrowedTo?: ClientModes | undefined;
/** A linked environment's thread this request is made for (`T3-Sender-Thread`). */
readonly senderThread?: OrchestrationV2LinkedSender | undefined;
/** An outside client's thread links go through this hosted app, as Copy thread's do. */
readonly hostedAppUrl?: string | undefined;
}
Expand All @@ -57,6 +60,13 @@ export interface McpClientCaller {
*/
export const MODE_LIMIT_HEADER = "t3-mode-limit";

/**
* `T3-Sender-Thread: <environmentId>/<threadId>/<title>`, each URI-encoded: the
* thread another environment makes a request for, so what it sends here can name
* it. Read only from a linked client.
*/
export const SENDER_THREAD_HEADER = "t3-sender-thread";

export interface ClientModes {
readonly runtimeMode: RuntimeMode;
readonly interactionMode: ProviderInteractionMode;
Expand Down
49 changes: 49 additions & 0 deletions apps/server/src/mcp/OrchestratorMcpService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3812,6 +3812,55 @@ describe("OrchestratorMcpService thread lookup", () => {
}),
);

it.effect("records the linked environment's thread that sent a message", () =>
Effect.gen(function* () {
const target = liveThreadShell(ThreadId.make("thread:target"));
const sender = {
environmentId: EnvironmentId.make("environment:a"),
threadId: ThreadId.make("thread:orc"),
title: "orc lead",
};
const linked = (senderThread?: typeof sender): McpInvocationScope => ({
...client,
client: {
sessionId: "link-a",
label: "A",
access: "full-access",
linked: true,
...(senderThread === undefined ? {} : { senderThread }),
},
});
const sendAs = (scope: McpInvocationScope) =>
Effect.gen(function* () {
const inputs: Array<ThreadManagementService.ThreadManagementSendInput> = [];
const sent = yield* Ref.make(0);
yield* Effect.gen(function* () {
const mcp = yield* OrchestratorMcpService.OrchestratorMcpService;
yield* mcp.sendToThread(scope, { threadId: target.id, message: "hello" });
}).pipe(
Effect.provide(
service([target], sent, {
getProjectThreadRecords: () => Effect.succeed(idleThreadProjection(target)),
sendToThread: (input) => {
inputs.push(input);
return Effect.succeed({
run: { id: RunId.make("run:sent"), status: "running" },
delivery: "started",
} as never);
},
}),
),
);
return inputs[0];
});

const fromLink = yield* sendAs(linked(sender));
assert.deepStrictEqual(fromLink?.linkedSender, sender);
assert.isUndefined(fromLink?.senderThreadId);
assert.isUndefined((yield* sendAs(linked()))?.linkedSender);
}),
);

it.effect("refuses waiting without orchestration capability before reading the target", () =>
Effect.gen(function* () {
const sent = yield* Ref.make(0);
Expand Down
4 changes: 4 additions & 0 deletions apps/server/src/mcp/OrchestratorMcpService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3144,6 +3144,10 @@ const make = Effect.gen(function* () {
}),
threadId: input.threadId,
...(parent === undefined ? {} : { senderThreadId: parent.thread.id }),
// A linked environment's agent has no thread here; the link names it.
...(parent === undefined && scope.client?.senderThread !== undefined
? { linkedSender: scope.client.senderThread }
: {}),
messageId,
text: input.message,
attachments: [],
Expand Down
3 changes: 3 additions & 0 deletions apps/server/src/orchestration-v2/EffectWorker.ts
Original file line number Diff line number Diff line change
Expand Up @@ -333,6 +333,9 @@ export const layerExecutor: Layer.Layer<
...(message.senderThreadId === undefined
? {}
: { senderThreadId: message.senderThreadId }),
...(message.linkedSender === undefined
? {}
: { linkedSender: message.linkedSender }),
})
.pipe(
// A disabled instance cannot retry this completed turn; its result stands.
Expand Down
15 changes: 15 additions & 0 deletions apps/server/src/orchestration-v2/Orchestrator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2643,6 +2643,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
...(queuedMessage.senderThreadId === undefined
? {}
: { senderThreadId: queuedMessage.senderThreadId }),
...(queuedMessage.linkedSender === undefined
? {}
: { linkedSender: queuedMessage.linkedSender }),
...(queuedMessage.idleCompaction === undefined
? {}
: { idleCompaction: queuedMessage.idleCompaction }),
Expand Down Expand Up @@ -5460,6 +5463,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
readonly creationSource: OrchestrationV2ConversationMessage["creationSource"];
readonly scheduledTaskId?: OrchestrationV2ConversationMessage["scheduledTaskId"];
readonly senderThreadId?: OrchestrationV2ConversationMessage["senderThreadId"];
readonly linkedSender?: OrchestrationV2ConversationMessage["linkedSender"];
readonly idleCompaction?: OrchestrationV2ConversationMessage["idleCompaction"];
readonly delegatedCompletion?: OrchestrationV2ConversationMessage["delegatedCompletion"];
readonly notification?: OrchestrationV2ConversationMessage["notification"];
Expand Down Expand Up @@ -5604,6 +5608,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
? {}
: { scheduledTaskId: input.scheduledTaskId }),
...(input.senderThreadId === undefined ? {} : { senderThreadId: input.senderThreadId }),
...(input.linkedSender === undefined ? {} : { linkedSender: input.linkedSender }),
...(input.idleCompaction === undefined ? {} : { idleCompaction: input.idleCompaction }),
id: input.messageId,
threadId: input.command.threadId,
Expand All @@ -5624,6 +5629,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
? {}
: { scheduledTaskId: input.scheduledTaskId }),
...(input.senderThreadId === undefined ? {} : { senderThreadId: input.senderThreadId }),
...(input.linkedSender === undefined ? {} : { linkedSender: input.linkedSender }),
...(input.idleCompaction === undefined ? {} : { idleCompaction: input.idleCompaction }),
id: idAllocator.derive.userTurnItem({ messageId: input.messageId }),
threadId: input.command.threadId,
Expand Down Expand Up @@ -6758,6 +6764,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
? {}
: { scheduledTaskId: command.scheduledTaskId }),
...(command.senderThreadId === undefined ? {} : { senderThreadId: command.senderThreadId }),
...(command.linkedSender === undefined ? {} : { linkedSender: command.linkedSender }),
...(command.idleCompaction === undefined ? {} : { idleCompaction: command.idleCompaction }),
forceRestart: dispatchMode.type === "restart_active",
});
Expand Down Expand Up @@ -7017,6 +7024,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
? {}
: { scheduledTaskId: command.scheduledTaskId }),
...(command.senderThreadId === undefined ? {} : { senderThreadId: command.senderThreadId }),
...(command.linkedSender === undefined ? {} : { linkedSender: command.linkedSender }),
...(command.idleCompaction === undefined ? {} : { idleCompaction: command.idleCompaction }),
id: command.messageId,
threadId: command.threadId,
Expand Down Expand Up @@ -7407,6 +7415,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
? {}
: { scheduledTaskId: command.scheduledTaskId }),
...(command.senderThreadId === undefined ? {} : { senderThreadId: command.senderThreadId }),
...(command.linkedSender === undefined ? {} : { linkedSender: command.linkedSender }),
...(command.idleCompaction === undefined ? {} : { idleCompaction: command.idleCompaction }),
id: command.messageId,
threadId: command.threadId,
Expand All @@ -7429,6 +7438,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
? {}
: { scheduledTaskId: command.scheduledTaskId }),
...(command.senderThreadId === undefined ? {} : { senderThreadId: command.senderThreadId }),
...(command.linkedSender === undefined ? {} : { linkedSender: command.linkedSender }),
...(command.idleCompaction === undefined ? {} : { idleCompaction: command.idleCompaction }),
id: idAllocator.derive.userTurnItem({ messageId: command.messageId }),
threadId: command.threadId,
Expand Down Expand Up @@ -8192,6 +8202,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
? {}
: { scheduledTaskId: command.scheduledTaskId }),
...(command.senderThreadId === undefined ? {} : { senderThreadId: command.senderThreadId }),
...(command.linkedSender === undefined ? {} : { linkedSender: command.linkedSender }),
...(command.idleCompaction === undefined ? {} : { idleCompaction: command.idleCompaction }),
id: command.messageId,
threadId: command.threadId,
Expand All @@ -8214,6 +8225,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
? {}
: { scheduledTaskId: command.scheduledTaskId }),
...(command.senderThreadId === undefined ? {} : { senderThreadId: command.senderThreadId }),
...(command.linkedSender === undefined ? {} : { linkedSender: command.linkedSender }),
...(command.idleCompaction === undefined ? {} : { idleCompaction: command.idleCompaction }),
id: idAllocator.derive.userTurnItem({ messageId: command.messageId }),
threadId: command.threadId,
Expand Down Expand Up @@ -10689,6 +10701,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
...(queuedMessage.senderThreadId === undefined
? {}
: { senderThreadId: queuedMessage.senderThreadId }),
...(queuedMessage.linkedSender === undefined
? {}
: { linkedSender: queuedMessage.linkedSender }),
...(queuedMessage.idleCompaction === undefined
? {}
: { idleCompaction: queuedMessage.idleCompaction }),
Expand Down
3 changes: 3 additions & 0 deletions apps/server/src/orchestration-v2/ThreadManagementService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import {
type OrchestrationV2ServerCommand,
type OrchestrationV2ConversationMessage,
type OrchestrationV2CreationSource,
type OrchestrationV2LinkedSender,
type OrchestrationV2Run,
type OrchestrationV2ThreadShellSnapshot,
type OrchestrationV2ThreadProjection,
Expand Down Expand Up @@ -143,6 +144,7 @@ export interface ThreadManagementSendInput {
readonly messageId: MessageId;
readonly scheduledTaskId?: ScheduledTaskId;
readonly senderThreadId?: ThreadId;
readonly linkedSender?: OrchestrationV2LinkedSender;
readonly text: string;
readonly attachments: ReadonlyArray<ChatAttachment>;
readonly modelSelection?: ModelSelection;
Expand Down Expand Up @@ -866,6 +868,7 @@ const make = Effect.gen(function* () {
messageId: input.messageId,
...(input.scheduledTaskId === undefined ? {} : { scheduledTaskId: input.scheduledTaskId }),
...(input.senderThreadId === undefined ? {} : { senderThreadId: input.senderThreadId }),
...(input.linkedSender === undefined ? {} : { linkedSender: input.linkedSender }),
text: input.text,
attachments: input.attachments,
...(input.modelSelection === undefined ? {} : { modelSelection: input.modelSelection }),
Expand Down
Loading
Loading