diff --git a/src/agent/agents/remoteEditorAgent.ts b/src/agent/agents/remoteEditorAgent.ts index fa9b1e79c9..d1ff7e134a 100644 --- a/src/agent/agents/remoteEditorAgent.ts +++ b/src/agent/agents/remoteEditorAgent.ts @@ -26,6 +26,7 @@ import { createComponentSearchTools } from "../tools/componentSearchTools"; import { createCsomTools } from "../tools/csomTools"; import { createDebugTools } from "../tools/debugTools"; import { createRunTools } from "../tools/runTools"; +import { createSessionResourceTools } from "../tools/sessionResourceTools"; import type { AgentContext } from "../types"; export interface RemoteAgentSpec { @@ -111,7 +112,9 @@ export function selectRemoteEditorTools( return registry.filter((toolDef) => allowed.has(toolDef.name)); } -function formatToolAvailabilitySection(tools: RemoteEditorTool[]): string { +function formatToolAvailabilitySection( + tools: readonly { name: string }[], +): string { const names = tools.map((toolDef) => `\`${toolDef.name}\``).join(", "); const available = names || "none"; return `\n\n## Available tools\n\nYou have exactly these tools: ${available}. Do not attempt workflow steps that require a tool outside this list. If a missing tool prevents the requested work, state that limitation clearly.`; @@ -150,7 +153,7 @@ function formatCurrentRunSection(context: AgentContext): string { function buildInstructions( session: AgentSession, spec: RemoteAgentSpec, - tools: RemoteEditorTool[], + tools: readonly { name: string }[], ): string { const base = session.context.mode === "runView" ? remoteRunPrompt : remoteEditorPrompt; @@ -173,7 +176,17 @@ export function buildRemoteEditorAgent( model, reasoningEffort: getInheritedReasoningEffort(session.aiConfig, model), }); - const tools = selectRemoteEditorTools(session, spec.tools); + // Session-resource tools are a host capability, not a server-granted CSOM + // tool, so they are always attached (outside the spawn allowlist) — mirroring + // the pi host, whose resource extension is likewise always registered. They + // appear only when the host wired a resource bridge for this turn. + const resourceTools = session.resources + ? createSessionResourceTools(session.resources).allTools + : []; + const tools = [ + ...selectRemoteEditorTools(session, spec.tools), + ...resourceTools, + ]; const effort = reasoningEffort(spec.thinkingDepth); const agent = new Agent({ name: "tangle-remote-editor", diff --git a/src/agent/createRemoteEnvWorkerApi.ts b/src/agent/createRemoteEnvWorkerApi.ts index 3c13a5859c..f0c059fb65 100644 --- a/src/agent/createRemoteEnvWorkerApi.ts +++ b/src/agent/createRemoteEnvWorkerApi.ts @@ -37,6 +37,7 @@ import { } from "./middleware/agentTrace"; import { recordTurnReasoning } from "./middleware/recordTurnReasoning"; import { createSession, type RecentPipelineRun } from "./session"; +import type { SessionResourceBridge } from "./sessionResourceBridge"; import { SkillsLoader } from "./skills/loader"; import type { ToolBridgeApi } from "./toolBridgeApi"; import type { AgentContext, StatusCallback } from "./types"; @@ -71,6 +72,7 @@ export interface RemoteEnvWorkerApi { runTurn( params: RemoteRunTurnParams, onStatus: StatusCallback, + resources?: SessionResourceBridge, ): Promise; abortAgent(agentId: string): void; killAgent(agentId: string): void; @@ -153,7 +155,7 @@ export function createRemoteEnvWorkerApi(): RemoteEnvWorkerApi { }); }, - async runTurn({ agentId, message, recentRuns }, onStatus) { + async runTurn({ agentId, message, recentRuns }, onStatus, resources) { const hosted = agents.get(agentId); if (!hosted) { throw new Error( @@ -185,6 +187,7 @@ export function createRemoteEnvWorkerApi(): RemoteEnvWorkerApi { recentRuns, context, componentCatalog: hosted.componentCatalog, + ...(resources ? { resources } : {}), }); // Once per spawn, not per turn: the allowlist is fixed when the agent is diff --git a/src/agent/prompts/remoteEditor.md b/src/agent/prompts/remoteEditor.md index 078929a989..ad7bf87bb7 100644 --- a/src/agent/prompts/remoteEditor.md +++ b/src/agent/prompts/remoteEditor.md @@ -36,6 +36,10 @@ Two things to know before calling it: Layout is not correctness: never use it to try to resolve a validation issue. +## Session resources + +The session has a shared resource catalog — files Prime, the user, or agents on other environments contributed (e.g. an uploaded `uploads/*.pipeline.yaml`). When a directive points at such a file, that is where it lives: `list_session_resources` shows the catalog, `read_session_resource` fetches one by its uri (e.g. `uploads/report.yaml`), and `put_session_resource` writes a file back into the session's `uploads/` folder for others to fetch. These are separate from the live canvas: reading a resource does not change the pipeline, and editing the pipeline does not touch the catalog. + ## Active subgraph context `get_pipeline_state` may include an `activeSubgraphPath` breadcrumb of subgraph task names from the root pipeline to whatever subgraph the user is viewing. Treat it as a hint about what the user cares about, but remember every CSOM mutation applies to the root spec. If a change targets an entity inside a nested subgraph, point that out before editing. diff --git a/src/agent/session.ts b/src/agent/session.ts index 780a3210fa..5adb88e548 100644 --- a/src/agent/session.ts +++ b/src/agent/session.ts @@ -6,6 +6,7 @@ import type { AiProviderConfig } from "@/types/aiProvider"; import type { ComponentCatalog } from "./componentCatalog"; import { createComponentCatalog } from "./componentCatalog"; import type { ProxyClient } from "./config"; +import type { SessionResourceBridge } from "./sessionResourceBridge"; import type { SkillsLoader } from "./skills/loader"; import type { ToolBridgeApi } from "./toolBridgeApi"; import type { AgentContext, AgentResponse, StatusCallback } from "./types"; @@ -29,6 +30,7 @@ export interface AgentSession { context: AgentContext; componentReferences: AgentResponse["componentReferences"]; componentCatalog: ComponentCatalog; + resources?: SessionResourceBridge; } export function createSession(params: { @@ -41,6 +43,7 @@ export function createSession(params: { emitStatus?: StatusCallback; recentRuns?: RecentPipelineRun[]; componentCatalog?: ComponentCatalog; + resources?: SessionResourceBridge; }): AgentSession { return { threadId: params.threadId, @@ -53,5 +56,6 @@ export function createSession(params: { context: params.context, componentReferences: {}, componentCatalog: params.componentCatalog ?? createComponentCatalog(), + ...(params.resources ? { resources: params.resources } : {}), }; } diff --git a/src/agent/sessionResourceBridge.ts b/src/agent/sessionResourceBridge.ts new file mode 100644 index 0000000000..93db9ba4a7 --- /dev/null +++ b/src/agent/sessionResourceBridge.ts @@ -0,0 +1,14 @@ +/** + * Main-thread access to the Tangent session's resource catalog, handed to the + * worker as a Comlink proxy per turn. The socket client that backs it lives in + * `remoteEnvHost.ts`; the worker's sub-agent reaches it only through this + * narrow surface, scoped to the turn's session. + */ +import type { RemoteResourceInput } from "@tangent/remote-subagent"; +import type { Resource } from "@tangent/shared/contracts.ts"; + +export interface SessionResourceBridge { + list(): Promise; + read(uri: string): Promise<{ content: string; contentType: string }>; + put(input: RemoteResourceInput): Promise; +} diff --git a/src/agent/tools/sessionResourceTools.ts b/src/agent/tools/sessionResourceTools.ts new file mode 100644 index 0000000000..43b06d5e90 --- /dev/null +++ b/src/agent/tools/sessionResourceTools.ts @@ -0,0 +1,101 @@ +/** + * Session-resource tools for the remote editor sub-agent. + * + * These mirror the pi host's session-resource extension so a browser sub-agent + * can see and fetch files Prime or agents on other environments contributed to + * the Tangent session (e.g. an uploaded `uploads/*.yaml`). They are host + * capabilities rather than CSOM mutations: each call reaches the main-thread + * socket client through the proxied `SessionResourceBridge`, scoped to the + * turn's session. + */ +import { tool } from "@openai/agents"; +import type { Resource } from "@tangent/shared/contracts.ts"; +import { z } from "zod"; + +import { decodeBase64Utf8 } from "@/utils/string"; + +import type { SessionResourceBridge } from "../sessionResourceBridge"; + +function renderResourceList(resources: Resource[]): string { + if (resources.length === 0) return "(no session resources)"; + return resources + .map( + (resource) => `- ${resource.name} (${resource.uri}) [${resource.kind}]`, + ) + .join("\n"); +} + +export function createSessionResourceTools(resources: SessionResourceBridge) { + const listSessionResources = tool({ + name: "list_session_resources", + description: + "List the files and resources catalogued in the Tangent session, including those contributed by Prime and agents on other environments. Returns each resource's name and uri.", + parameters: z.object({}), + execute: async () => { + const catalog = await resources.list(); + return `## Session resources\n\n${renderResourceList(catalog)}`; + }, + }); + + const readSessionResource = tool({ + name: "read_session_resource", + description: + "Read a Tangent session resource's contents by its uri (e.g. 'uploads/report.yaml'), including content another environment contributed. Returns the decoded text.", + parameters: z.object({ + uri: z + .string() + .describe( + "Session-root-relative resource uri, e.g. 'uploads/report.yaml'.", + ), + }), + execute: async ({ uri }) => { + const { content } = await resources.read(uri); + return decodeBase64Utf8(content); + }, + }); + + const putSessionResource = tool({ + name: "put_session_resource", + description: + "Write a file into the Tangent session's uploads/ folder so Prime and other agents (including those on other environments) can fetch it. Binary content must be base64 with encoding 'base64'.", + parameters: z.object({ + path: z + .string() + .describe("Destination path inside the session's uploads/ folder."), + content: z.string().describe("File contents (text, or base64)."), + encoding: z + .enum(["utf8", "base64"]) + .nullable() + .optional() + .describe("How content is decoded. Defaults to utf8."), + name: z + .string() + .nullable() + .optional() + .describe("Catalog display name; defaults to the path basename."), + contentType: z + .string() + .nullable() + .optional() + .describe("MIME type recorded in the catalog meta."), + }), + execute: async ({ path, content, encoding, name, contentType }) => { + const stored = await resources.put({ + kind: "file", + path, + content, + encoding: encoding ?? "utf8", + ...(name ? { name } : {}), + ...(contentType ? { meta: { contentType } } : {}), + }); + return `Added session resource ${stored.uri}`; + }, + }); + + return { + listSessionResources, + readSessionResource, + putSessionResource, + allTools: [listSessionResources, readSessionResource, putSessionResource], + }; +} diff --git a/src/routes/v2/pages/Tangent/components/TangentProjectAgentProvider.tsx b/src/routes/v2/pages/Tangent/components/TangentProjectAgentProvider.tsx index 6b31f93753..30160d9a2e 100644 --- a/src/routes/v2/pages/Tangent/components/TangentProjectAgentProvider.tsx +++ b/src/routes/v2/pages/Tangent/components/TangentProjectAgentProvider.tsx @@ -33,7 +33,10 @@ import { copyRunToPipeline } from "@/services/pipelineRunService"; import { usePipelineStorage } from "@/services/pipelineStorage/PipelineStorageProvider"; import type { PipelineStorageService } from "@/services/pipelineStorage/PipelineStorageService"; import type { RemotePipelineRef } from "@/services/pipelineStorage/types"; -import { createAndAttachPipeline } from "@/services/projects/createAndAttachPipeline"; +import { + createAndAttachPipeline, + importAndAttachPipeline, +} from "@/services/projects/createAndAttachPipeline"; import { pipelineFileTarget, remotePipelineResourceInput, @@ -53,6 +56,7 @@ import { extractCanonicalName } from "@/utils/canonicalPipelineName"; import { isValidComponentSpec } from "@/utils/componentSpec"; import { getInitialName } from "@/utils/getComponentName"; import { extractCloneableTaskArguments } from "@/utils/nodes/taskArguments"; +import { decodeBase64Utf8 } from "@/utils/string"; interface TangentProjectAgentProviderProps { sessionId: string | undefined; @@ -131,6 +135,9 @@ export function TangentProjectAgentProvider({ sessionId ? { projectId: store.projectId, sessionId } : undefined, ); const workerRef = useRef | null>(null); + const readSessionResourceRef = useRef< + ((uri: string) => Promise) | null + >(null); // A project-level backend bridge for the run inspect tools: read-only run and // execution fetches that don't need any one tab's canvas spec, so Prime can @@ -230,6 +237,25 @@ export function TangentProjectAgentProvider({ target: pipelineFileTarget(file), }; }, + importPipeline: async (yaml, name) => { + const file = await importAndAttachPipeline({ + storage, + yaml, + name, + attach: (input) => createProjectResource(store.projectId, input), + }); + await refreshProjectResources(); + return { pipelineName: file.storageKey, fileId: file.id }; + }, + readSessionResource: async (uri) => { + const read = readSessionResourceRef.current; + if (!read) { + throw new Error( + "The session environment is not connected, so its resources cannot be read yet.", + ); + } + return read(uri); + }, clonePipeline: async (runId) => { const { pipelineName, remote } = await clonePipelineFromRun( storage, @@ -298,6 +324,11 @@ export function TangentProjectAgentProvider({ metadata: describeBrowserEnv({ mode: "editor" }), }); + readSessionResourceRef.current = async (uri) => { + const { content } = await host.readResource(uri); + return decodeBase64Utf8(content); + }; + const stop = connectRemoteEnvWithRefresh({ host, baseUrl, @@ -314,6 +345,7 @@ export function TangentProjectAgentProvider({ host.disconnect(); worker.terminate(); workerRef.current = null; + readSessionResourceRef.current = null; }; }, [sessionId, tools, routingBridge, agentTargets, baseUrl]); diff --git a/src/routes/v2/pages/Tangent/services/connectRemoteEnvWithRefresh.test.ts b/src/routes/v2/pages/Tangent/services/connectRemoteEnvWithRefresh.test.ts index f3219ef442..3850583cc1 100644 --- a/src/routes/v2/pages/Tangent/services/connectRemoteEnvWithRefresh.test.ts +++ b/src/routes/v2/pages/Tangent/services/connectRemoteEnvWithRefresh.test.ts @@ -12,6 +12,7 @@ function makeHost(): RemoteEnvHost { return { connect: vi.fn().mockResolvedValue(undefined), disconnect: vi.fn(), + readResource: vi.fn(), }; } diff --git a/src/routes/v2/pages/Tangent/services/createWorkareaRemoteTools.test.ts b/src/routes/v2/pages/Tangent/services/createWorkareaRemoteTools.test.ts index c35918493e..f5320b6cb3 100644 --- a/src/routes/v2/pages/Tangent/services/createWorkareaRemoteTools.test.ts +++ b/src/routes/v2/pages/Tangent/services/createWorkareaRemoteTools.test.ts @@ -61,6 +61,8 @@ function makeDeps(overrides: Partial = {}): WorkareaToolDeps { waitForEnvironment: vi.fn().mockResolvedValue(undefined), runInspect: makeRunInspect(), createPipeline: vi.fn(), + importPipeline: vi.fn(), + readSessionResource: vi.fn(), clonePipeline: vi.fn(), refreshResources: vi.fn().mockResolvedValue(undefined), renameProject: vi.fn().mockResolvedValue({ renamed: true }), @@ -396,6 +398,102 @@ describe("createWorkareaRemoteTools", () => { expect(createPipeline).not.toHaveBeenCalled(); }); + it("import_pipeline imports the YAML, opens it, and reports the tab it can spawn into", async () => { + const importPipeline = vi + .fn() + .mockResolvedValue({ pipelineName: "Imported DAG", fileId: "file-7" }); + const tab = { ...pipelineTab("tab-imp", "file-7"), title: "Imported DAG" }; + const openTarget = vi.fn().mockResolvedValue(tab); + const tools = createWorkareaRemoteTools(() => + makeDeps({ + importPipeline, + openTarget, + getTabs: () => [tab], + getActiveTabId: () => "tab-imp", + waitForEnvironment: vi.fn().mockResolvedValue("env-7"), + }), + ); + + const result = await tools.import_pipeline.execute({ + yaml: "name: Imported DAG\nimplementation:\n graph:\n tasks: {}\n", + name: "Imported DAG", + }); + + expect(importPipeline).toHaveBeenCalledWith( + "name: Imported DAG\nimplementation:\n graph:\n tasks: {}\n", + "Imported DAG", + ); + expect(openTarget).toHaveBeenCalledWith( + { type: "pipeline", identity: "id/file-7" }, + "Imported DAG", + ); + expect(result).toEqual({ + id: "tab-imp", + kind: "pipeline", + name: "Imported DAG", + title: "Imported DAG", + target: "pipeline://id/file-7", + active: true, + environmentId: "env-7", + ready: true, + }); + }); + + it("import_pipeline reads a session resource handle and imports its YAML", async () => { + const yaml = + "name: From Resource\nimplementation:\n graph:\n tasks: {}\n"; + const readSessionResource = vi.fn().mockResolvedValue(yaml); + const importPipeline = vi + .fn() + .mockResolvedValue({ pipelineName: "From Resource", fileId: "file-r" }); + const tab = { ...pipelineTab("tab-r", "file-r"), title: "From Resource" }; + const tools = createWorkareaRemoteTools(() => + makeDeps({ + readSessionResource, + importPipeline, + openTarget: vi.fn().mockResolvedValue(tab), + getTabs: () => [tab], + getActiveTabId: () => "tab-r", + waitForEnvironment: vi.fn().mockResolvedValue("env-r"), + }), + ); + + const result = await tools.import_pipeline.execute({ + resourceUri: "uploads/pipeline.yaml", + }); + + expect(readSessionResource).toHaveBeenCalledWith("uploads/pipeline.yaml"); + expect(importPipeline).toHaveBeenCalledWith(yaml, undefined); + expect(result).toMatchObject({ + id: "tab-r", + name: "From Resource", + ready: true, + }); + }); + + it("import_pipeline refuses when neither a resource handle nor YAML is given", async () => { + const importPipeline = vi.fn(); + const readSessionResource = vi.fn(); + const tools = createWorkareaRemoteTools(() => + makeDeps({ importPipeline, readSessionResource }), + ); + + await expect(tools.import_pipeline.execute({})).rejects.toThrow( + /resourceUri/, + ); + await expect( + tools.import_pipeline.execute({ yaml: " " }), + ).rejects.toThrow(/non-empty string/); + await expect( + tools.import_pipeline.execute({ + resourceUri: "uploads/a.yaml", + yaml: "name: x", + }), + ).rejects.toThrow(/not both/); + expect(importPipeline).not.toHaveBeenCalled(); + expect(readSessionResource).not.toHaveBeenCalled(); + }); + it("clone_pipeline clones the resolved run tab and returns the new pipeline target", async () => { const clonePipeline = vi .fn() diff --git a/src/routes/v2/pages/Tangent/services/createWorkareaRemoteTools.ts b/src/routes/v2/pages/Tangent/services/createWorkareaRemoteTools.ts index 0b0ba0707e..78f8180c5e 100644 --- a/src/routes/v2/pages/Tangent/services/createWorkareaRemoteTools.ts +++ b/src/routes/v2/pages/Tangent/services/createWorkareaRemoteTools.ts @@ -50,6 +50,11 @@ export interface WorkareaToolDeps { fileId: string; target?: WorkareaTarget; }>; + importPipeline: ( + yaml: string, + name?: string, + ) => Promise<{ pipelineName: string; fileId: string }>; + readSessionResource: (uri: string) => Promise; clonePipeline: ( runId: string, ) => Promise<{ pipelineName: string; target?: WorkareaTarget }>; @@ -139,6 +144,33 @@ function requireName(args: unknown): string { return name; } +function optionalYaml(args: unknown): string | undefined { + if (!isRecord(args) || !("yaml" in args) || args.yaml === undefined) { + return undefined; + } + if (typeof args.yaml !== "string" || args.yaml.trim().length === 0) { + throw new Error("`yaml` must be a non-empty string when provided."); + } + return args.yaml; +} + +function optionalResourceUri(args: unknown): string | undefined { + if ( + !isRecord(args) || + !("resourceUri" in args) || + args.resourceUri === undefined + ) { + return undefined; + } + if ( + typeof args.resourceUri !== "string" || + args.resourceUri.trim().length === 0 + ) { + throw new Error("`resourceUri` must be a non-empty string when provided."); + } + return args.resourceUri.trim(); +} + function optionalRunId(args: unknown): string | undefined { if (!isRecord(args) || !("runId" in args)) return undefined; if (typeof args.runId !== "string" || args.runId.trim().length === 0) { @@ -291,6 +323,68 @@ export function createWorkareaRemoteTools( return { ...tab, name: pipelineName }; }, }, + import_pipeline: { + description: + "Import a complete pipeline and open it in the Dynamic Workarea, rather " + + "than rebuilding it node by node. Give it either `resourceUri` — the " + + "handle of a Tangent session resource holding the pipeline YAML, which " + + "is how you bring in a pipeline another agent contributed — or inline " + + "`yaml`. The content must be a graph pipeline spec. `name` is optional " + + "and is made unique if it is already taken; otherwise the name from the " + + "YAML is used. The pipeline is attached to the project automatically " + + "(it shows up in Resources). Returns the tab summary plus the pipeline " + + "`name` and its `pipeline://id/` `target`, including the " + + "`environmentId` and `ready` flag for that tab's sub-agent host, so a " + + "spawn can follow straight on.", + inputSchema: { + type: "object", + properties: { + resourceUri: { + type: "string", + description: + "Handle of a session resource holding the pipeline YAML (its `uri`, e.g. `uploads/pipeline.yaml`). Preferred for content contributed to the session.", + }, + yaml: { + type: "string", + description: + "Inline pipeline YAML content, as an alternative to `resourceUri`.", + }, + name: { + type: "string", + description: + "Optional name for the imported pipeline. Defaults to the name in the YAML.", + }, + }, + }, + execute: async (args) => { + const resourceUri = optionalResourceUri(args); + const inlineYaml = optionalYaml(args); + if (resourceUri && inlineYaml) { + throw new Error("Provide either `resourceUri` or `yaml`, not both."); + } + const deps = getDeps(); + let pipelineYaml: string; + if (resourceUri) { + pipelineYaml = await deps.readSessionResource(resourceUri); + } else if (inlineYaml) { + pipelineYaml = inlineYaml; + } else { + throw new Error( + "Provide a `resourceUri` to a session resource, or inline `yaml`.", + ); + } + const { pipelineName, fileId } = await deps.importPipeline( + pipelineYaml, + optionalName(args), + ); + const target: WorkareaTarget = { + type: "pipeline", + identity: idIdentity(fileId), + }; + const tab = await openAndSummarize(getDeps, target, pipelineName); + return { ...tab, name: pipelineName }; + }, + }, clone_pipeline: { description: "Clone the pipeline behind a failed run into a new editable " + diff --git a/src/routes/v2/pages/Tangent/services/remoteEnvHost.ts b/src/routes/v2/pages/Tangent/services/remoteEnvHost.ts index 7813990af1..af798c717c 100644 --- a/src/routes/v2/pages/Tangent/services/remoteEnvHost.ts +++ b/src/routes/v2/pages/Tangent/services/remoteEnvHost.ts @@ -31,6 +31,7 @@ import type { Remote } from "comlink"; import { proxy } from "comlink"; import type { RemoteEnvWorkerApi } from "@/agent/createRemoteEnvWorkerApi"; +import type { SessionResourceBridge } from "@/agent/sessionResourceBridge"; import type { AgentContext } from "@/agent/types"; import type { AgentTargetRouter } from "@/routes/v2/pages/Tangent/services/createActiveTabRoutingBridge"; import { getTangentSocketConfig } from "@/routes/v2/pages/Tangent/services/socketConfig"; @@ -185,6 +186,7 @@ export type RemoteEnvHostOptions = RemoteEnvHostBaseOptions & export interface RemoteEnvHost { connect(token: string, environmentId: string): Promise; disconnect(): void; + readResource(uri: string): Promise<{ content: string; contentType: string }>; } interface AgentLifecycle { @@ -273,9 +275,23 @@ export function createRemoteEnvHost( emitActivity(command, lifecycle, status.text), ); agentTargets?.pinTurn(agentId); + // The socket client lives here, so the sub-agent reads the session's + // resource catalog through this per-turn bridge rather than holding the + // connection in the worker. Scoped to the turn's session and agent, it + // mirrors the pi host's loopback resource API. + const turnClient = client; + const resources: SessionResourceBridge | undefined = turnClient + ? { + list: () => turnClient.listResources(turnSessionId), + read: (uri) => turnClient.readResource(turnSessionId, uri), + put: (input) => + turnClient.addResource(turnSessionId, agentId, input), + } + : undefined; const { answer } = await worker.runTurn( { agentId, message: text }, onStatus, + resources ? proxy(resources) : undefined, ); if (!isCurrent(agentId, lifecycle)) return; // The finalized `end` (with `content` + echoed `runId`) is the whole @@ -479,5 +495,21 @@ export function createRemoteEnvHost( client?.disconnect(); client = null; }, + + // A session-scoped read so a host tool can pull a resource by its handle + // (e.g. a pipeline another agent contributed) without holding the socket + // itself. Unlike the per-turn worker bridge this needs no agentId: reading + // the session catalog is not attributed to any one sub-agent. + readResource(uri) { + if (!client) { + throw new Error( + "Cannot read a session resource before the environment connects.", + ); + } + if (sessionId === undefined) { + throw new Error("This environment has no session to read resources."); + } + return client.readResource(sessionId, uri); + }, }; } diff --git a/src/services/projects/createAndAttachPipeline.ts b/src/services/projects/createAndAttachPipeline.ts index d93e8ab7fd..3e04bfeaf3 100644 --- a/src/services/projects/createAndAttachPipeline.ts +++ b/src/services/projects/createAndAttachPipeline.ts @@ -3,6 +3,8 @@ import { availablePipelineName } from "@/services/localPipelines/localPipelinesS import { getPipelineCreationBackendUrl } from "@/services/pipelineStorage/pipelineCreation"; import type { PipelineFile } from "@/services/pipelineStorage/PipelineFile"; import type { PipelineStorageService } from "@/services/pipelineStorage/PipelineStorageService"; +import { isGraphImplementation } from "@/utils/componentSpec"; +import { componentSpecFromYaml, componentSpecToYaml } from "@/utils/yaml"; import { pipelineFileResourceInput, @@ -10,17 +12,47 @@ import { } from "./pipelineResource"; import type { CreateResourceInput } from "./types"; +type Attach = (input: CreateResourceInput) => Promise; + interface CreateAndAttachOptions { storage: PipelineStorageService; name?: string; provisionalName?: boolean; - attach: (input: CreateResourceInput) => Promise; + attach: Attach; +} + +interface ImportAndAttachOptions { + storage: PipelineStorageService; + yaml: string; + name?: string; + attach: Attach; } /** * Remote drafts are attached on their first save. Local creation attaches * immediately and removes the empty file if the project refuses it. */ +async function attachOrCleanup( + file: PipelineFile, + attach: Attach, +): Promise { + try { + await attach(pipelineFileResourceInput(file)); + } catch (error) { + await file + .deleteFile() + .catch((cleanupError: unknown) => + console.error( + "Failed to remove the pipeline whose project attachment was refused", + cleanupError, + ), + ); + throw error; + } + + return file; +} + export async function createAndAttachPipeline({ storage, name, @@ -42,19 +74,31 @@ export async function createAndAttachPipeline({ // The driver awaits attachment, so failures remain retryable through autosave. if (file.isDraft) return file; - try { - await attach(pipelineFileResourceInput(file)); - } catch (error) { - await file - .deleteFile() - .catch((cleanupError: unknown) => - console.error( - "Failed to remove the pipeline whose project attachment was refused", - cleanupError, - ), - ); - throw error; + return attachOrCleanup(file, attach); +} + +export async function importAndAttachPipeline({ + storage, + yaml, + name, + attach, +}: ImportAndAttachOptions): Promise { + const componentSpec = componentSpecFromYaml(yaml); + if (!isGraphImplementation(componentSpec.implementation)) { + throw new Error( + "This does not look like a pipeline: its implementation is not a graph.", + ); } - return file; + const storageKey = await availablePipelineName( + name ?? componentSpec.name ?? "Imported Pipeline", + ); + componentSpec.name = storageKey; + + const file = await storage.rootFolder.addFile( + storageKey, + componentSpecToYaml(componentSpec), + ); + + return attachOrCleanup(file, attach); } diff --git a/src/utils/string.ts b/src/utils/string.ts index ca6075aeac..29baa1bdec 100644 --- a/src/utils/string.ts +++ b/src/utils/string.ts @@ -115,6 +115,12 @@ export function getStringFromData(data: string | ArrayBuffer): string { return data; } +export function decodeBase64Utf8(base64: string): string { + const binary = atob(base64); + const bytes = Uint8Array.from(binary, (char) => char.charCodeAt(0)); + return new TextDecoder().decode(bytes); +} + export const truncate = ( str: string, maxLength: number,