Add the Space upstream relay

src/api/v2/agents/handlers/join-status.ts

@@ -19,7 +19,7 @@ const paramsSchema = z.object({
 // Optional dev-only variant routing hint. A malformed value (array, blank,
 // over-long) parses away to undefined and the poll falls back to the default
 // worker rather than 400ing — the status read still works.
-const querySchema = z.object({
+export const joinStatusQuerySchema = z.object({
   variantId: z.string().trim().min(1).max(64).optional(),
 });
 
@@ -85,7 +85,7 @@ export async function joinStatusHandler(req: Request, res: Response) {
   // the right runtime. Re-resolve the variant's ephemeral origin (dev-only, live +
   // allowlisted); anything else falls back to the default worker.
   let assistantBaseUrl = assistantApiUrl.replace(/\/+$/, "");
-  const variantId = querySchema.safeParse(req.query).data?.variantId;
+  const variantId = joinStatusQuerySchema.safeParse(req.query).data?.variantId;
   if (variantId && XMTP_ENV !== "production") {
     const origin = await resolveVariantWorkerOrigin(variantId);
     if (origin) assistantBaseUrl = origin;

src/api/v2/conversations/conversations.router.ts

@@ -1,19 +1,31 @@
 import { Router } from "express";
 import { requireAccount } from "@/middleware/auth";
-import { agentParticipationLimiter } from "@/middleware/rateLimit";
+import {
+  agentParticipationLimiter,
+  spaceUpstreamLimiter,
+} from "@/middleware/rateLimit";
 import { conversationAbilitiesGetHandler } from "./handlers/abilities-get";
 import { conversationAbilityDeleteHandler } from "./handlers/ability-delete";
 import { conversationAbilityPutHandler } from "./handlers/ability-put";
 import {
   getParticipationHandler,
   participationHandler,
 } from "./handlers/participation";
+import { spaceUpstreamHandler } from "./handlers/space-upstream";
 
 // /v2/conversations — conversation-scoped surfaces. Mounted behind
 // authMiddleware in src/api/v2/index.ts; every route here applies
 // requireAccount itself. conversationId is the opaque XMTP string (no
 // Conversation table).
 export const conversationsRouter = Router();
+export const conversationsDebugRouter = Router();
+
+conversationsDebugRouter.post(
+  "/:conversationId/debug/space-upstream",
+  spaceUpstreamLimiter,
+  requireAccount,
+  spaceUpstreamHandler,
+);
 
 // How much the agents in this conversation may speak. `requireAccount` for the
 // same reason as /agents/join: an account-less JWT is an authorization failure,

src/api/v2/conversations/handlers/space-upstream.ts

@@ -0,0 +1,242 @@
+import type { Request, Response } from "express";
+import { z } from "zod";
+import {
+  getAssistantApiKey,
+  getAssistantApiUrl,
+} from "@/api/v2/agents/handlers/assistant-config";
+import { joinStatusQuerySchema } from "@/api/v2/agents/handlers/join-status";
+import { resolveVariantWorkerOrigin } from "@/api/v2/agents/lib/variant-routing";
+
+export const SPACE_UPSTREAM_FETCH_TIMEOUT_MS = 50_000;
+const ERROR_BODY_LOG_LIMIT = 200;
+
+const conversationIdSchema = z
+  .string()
+  .regex(/^[0-9A-Za-z_-]{1,128}$/, "Invalid conversationId");
+
+const paramsSchema = z.object({
+  conversationId: conversationIdSchema,
+});
+
+const resultCountsSchema = {
+  wrote: z.number().int().nonnegative(),
+  deleted: z.number().int().nonnegative(),
+  refusedCount: z.number().int().nonnegative(),
+};
+
+export const spaceUpstreamResultSchema = z.discriminatedUnion("outcome", [
+  z
+    .object({
+      conversationId: conversationIdSchema,
+      outcome: z.literal("pull_request"),
+      prUrl: z.string().url(),
+      prNumber: z.number().int().positive(),
+      branch: z.string().min(1),
+      commitSha: z.string().min(1),
+      forkCommitSha: z.string().min(1),
+      ...resultCountsSchema,
+    })
+    .strict(),
+  z
+    .object({
+      conversationId: conversationIdSchema,
+      outcome: z.literal("unchanged"),
+      forkCommitSha: z.string().min(1),
+      ...resultCountsSchema,
+    })
+    .strict(),
+]);
+
+const upstreamErrorSchema = z
+  .object({
+    error: z.string().min(1).max(500),
+    code: z.string().min(1).max(64),
+  })
+  .strict();
+
+type PublicError = {
+  status: number;
+  code: string;
+  error: string;
+};
+
+const ERRORS = {
+  INVALID_REQUEST: {
+    status: 400,
+    code: "INVALID_REQUEST",
+    error: "Invalid Space PR proposal request",
+  },
+  VARIANT_UNAVAILABLE: {
+    status: 409,
+    code: "VARIANT_UNAVAILABLE",
+    error: "The selected agent variant is unavailable",
+  },
+  SPACE_NOT_FOUND: {
+    status: 404,
+    code: "SPACE_NOT_FOUND",
+    error: "No Space was found for this conversation",
+  },
+  SPACE_REPOSITORY_UNAVAILABLE: {
+    status: 409,
+    code: "SPACE_REPOSITORY_UNAVAILABLE",
+    error: "This Space does not have a repository",
+  },
+  SPACE_UPSTREAM_NOT_ARMED: {
+    status: 503,
+    code: "SPACE_UPSTREAM_NOT_ARMED",
+    error: "The selected Space deployment is not armed for PR proposals",
+  },
+  SPACE_UPSTREAM_UNAVAILABLE: {
+    status: 503,
+    code: "SPACE_UPSTREAM_UNAVAILABLE",
+    error: "Space PR proposals are unavailable",
+  },
+  SPACE_UPSTREAM_REFUSED: {
+    status: 422,
+    code: "SPACE_UPSTREAM_REFUSED",
+    error: "The Space changes could not be proposed safely",
+  },
+  SPACE_UPSTREAM_GITHUB_FAILED: {
+    status: 502,
+    code: "SPACE_UPSTREAM_GITHUB_FAILED",
+    error: "GitHub rejected the Space PR proposal; please try again",
+  },
+  SPACE_UPSTREAM_FAILED: {
+    status: 502,
+    code: "SPACE_UPSTREAM_FAILED",
+    error: "The Space PR proposal failed",
+  },
+  SPACE_UPSTREAM_TIMEOUT: {
+    status: 504,
+    code: "SPACE_UPSTREAM_TIMEOUT",
+    error: "The Space PR proposal timed out",
+  },
+} as const satisfies Record<string, PublicError>;
+
+function sendError(res: Response, value: PublicError): void {
+  const { status, ...body } = value;
+  res.status(status).json(body);
+}
+
+function translateUpstreamError(status: number, raw: unknown): PublicError {
+  const parsed = upstreamErrorSchema.safeParse(raw);
+  if (!parsed.success) return ERRORS.SPACE_UPSTREAM_FAILED;
+
+  const { code, error } = parsed.data;
+  if (status === 403 && code === "space_upstream_not_armed") {
+    return ERRORS.SPACE_UPSTREAM_NOT_ARMED;
+  }
+  if (status === 404 && code === "space_not_found") {
+    return ERRORS.SPACE_NOT_FOUND;
+  }
+  if (status === 409 && code === "space_repository_unavailable") {
+    return ERRORS.SPACE_REPOSITORY_UNAVAILABLE;
+  }
+  if (status === 503 && code === "space_repository_provider_unavailable") {
+    return ERRORS.SPACE_UPSTREAM_UNAVAILABLE;
+  }
+  if (status === 422 && code === "space_upstream_refused") {
+    return { ...ERRORS.SPACE_UPSTREAM_REFUSED, error };
+  }
+  if (status === 502 && code === "space_upstream_github_failed") {
+    return ERRORS.SPACE_UPSTREAM_GITHUB_FAILED;
+  }
+  if (status === 502 && code === "space_upstream_failed") {
+    return ERRORS.SPACE_UPSTREAM_FAILED;
+  }
+  if (status === 504 && code === "space_upstream_timeout") {
+    return ERRORS.SPACE_UPSTREAM_TIMEOUT;
+  }
+  return ERRORS.SPACE_UPSTREAM_FAILED;
+}
+
+export async function spaceUpstreamHandler(req: Request, res: Response) {
+  const parsedParams = paramsSchema.safeParse(req.params);
+  const parsedQuery = joinStatusQuerySchema.safeParse(req.query);
+  if (!parsedParams.success || !parsedQuery.success) {
+    sendError(res, ERRORS.INVALID_REQUEST);
+    return;
+  }
+
+  const conversationId = parsedParams.data.conversationId.toLowerCase();
+  const variantId = parsedQuery.data.variantId;
+
+  let assistantOrigin: string;
+  if (variantId !== undefined) {
+    const resolvedOrigin = await resolveVariantWorkerOrigin(variantId);
+    if (!resolvedOrigin) {
+      sendError(res, ERRORS.VARIANT_UNAVAILABLE);
+      return;
+    }
+    assistantOrigin = resolvedOrigin;
+  } else {
+    assistantOrigin = getAssistantApiUrl();
+  }
+
+  const assistantApiKey = getAssistantApiKey().trim();
+  const assistantBaseUrl = assistantOrigin.trim().replace(/\/+$/, "");
+  if (!assistantApiKey || !assistantBaseUrl) {
+    req.log.error("Space upstream Worker is not configured");
+    sendError(res, ERRORS.SPACE_UPSTREAM_UNAVAILABLE);
+    return;
+  }
+
+  try {
+    const upstream = await fetch(
+      `${assistantBaseUrl}/api/conversations/${encodeURIComponent(conversationId)}/space-upstream`,
+      {
+        method: "POST",
+        headers: { Authorization: `Bearer ${assistantApiKey}` },
+        signal: AbortSignal.timeout(SPACE_UPSTREAM_FETCH_TIMEOUT_MS),
+      },
+    );
+
+    if (!upstream.ok) {
+      const text = await upstream.text();
+      const bodyPreview = text.substring(0, ERROR_BODY_LOG_LIMIT);
+      req.log.error(
+        { status: upstream.status, bodyPreview },
+        "Space upstream Worker request failed",
+      );
+
+      let raw: unknown;
+      try {
+        raw = JSON.parse(text);
+      } catch {
+        raw = null;
+      }
+      sendError(res, translateUpstreamError(upstream.status, raw));
+      return;
+    }
+
+    let raw: unknown;
+    try {
+      raw = await upstream.json();
+    } catch {
+      raw = null;
+    }
+    const result = spaceUpstreamResultSchema.safeParse(raw);
+    if (!result.success) {
+      req.log.error(
+        { issues: result.error.issues },
+        "Invalid Space upstream Worker response",
+      );
+      sendError(res, ERRORS.SPACE_UPSTREAM_FAILED);
+      return;
+    }
+
+    res.status(200).json(result.data);
+  } catch (error) {
+    if (error instanceof DOMException && error.name === "TimeoutError") {
+      req.log.error("Space upstream Worker request timed out");
+      sendError(res, ERRORS.SPACE_UPSTREAM_TIMEOUT);
+      return;
+    }
+
+    req.log.error(
+      { error, stack: error instanceof Error ? error.stack : undefined },
+      "Space upstream Worker request failed",
+    );
+    sendError(res, ERRORS.SPACE_UPSTREAM_FAILED);
+  }
+}

src/api/v2/index.ts

@@ -40,7 +40,10 @@ import { composioRouter } from "./composio/composio.router";
 import { connectionsRouter } from "./connections/connections.router";
 import { actionsGetHandler } from "./connections/handlers/actions-get";
 import { servicesGetHandler } from "./connections/handlers/services-get";
-import { conversationsRouter } from "./conversations/conversations.router";
+import {
+  conversationsDebugRouter,
+  conversationsRouter,
+} from "./conversations/conversations.router";
 import { creditsAdminRouter } from "./credits-admin/credits-admin.router";
 import { dailyRefillRouter } from "./credits/daily.router";
 import { devRouter } from "./dev/dev.router";
@@ -61,6 +64,7 @@ const v2Router = Router();
 // /dev is a non-production test surface; keep it gated.
 if (process.env.XMTP_ENV !== "production") {
   v2Router.use("/dev", devAuthMiddleware, devRouter);
+  v2Router.use("/conversations", authMiddleware, conversationsDebugRouter);
 }
 
 v2Router.use("/agent-prompt-hints", agentPromptHintsRouter);

src/middleware/rateLimit.ts

@@ -57,6 +57,20 @@ export const agentParticipationLimiter = rateLimit({
   },
 });
 
+// Space-to-starter proposals can update a GitHub branch and draft pull request,
+// so keep retries bounded independently of the cheaper participation controls.
+export const spaceUpstreamLimiter = rateLimit({
+  windowMs: 5 * 60 * 1000,
+  limit: 10,
+  keyGenerator: (req) => req.ip || "unknown",
+  legacyHeaders: false,
+  standardHeaders: "draft-8",
+  message: {
+    code: "RATE_LIMITED",
+    error: "Too many Space PR proposals; retry shortly",
+  },
+});
+
 // Rate limiting for asset renewal endpoint (10 batch requests per hour per device)
 export const assetRenewalLimiter = rateLimit({
   windowMs: 60 * 60 * 1000, // 1 hour

tests/space-upstream-production.test.ts

@@ -0,0 +1,33 @@
+import express from "express";
+import request from "supertest";
+import { afterEach, describe, expect, test, vi } from "vitest";
+
+const originalXmtpEnv = process.env.XMTP_ENV;
+
+afterEach(() => {
+  process.env.XMTP_ENV = originalXmtpEnv;
+  vi.resetModules();
+});
+
+describe("Space upstream production mount guard", () => {
+  test("does not mount the debug route in production", async () => {
+    process.env.XMTP_ENV = "production";
+    vi.resetModules();
+    const { default: v2Router } = await import("@/api/v2");
+    const { pinoMiddleware } = await import("@/middleware/pino");
+    const { createJwtToken, validateJWTKeys } = await import("@/utils/jwt");
+    await validateJWTKeys();
+    const token = await createJwtToken({
+      deviceId: "production-mount-test",
+      accountId: "11111111-1111-4111-8111-111111111111",
+    });
+    const app = express();
+    app.use(pinoMiddleware);
+    app.use("/api/v2", v2Router);
+
+    const response = await request(app)
+      .post("/api/v2/conversations/conversation_abc/debug/space-upstream")
+      .set("X-Convos-AuthToken", token);
+    expect(response.status).toBe(404);
+  });
+});