Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| fd94163e00 | |||
| 231dfcf523 | |||
| e740ffe23f |
@@ -2,11 +2,8 @@
|
||||
|
||||
// 04 · Chat — message thread + composer + sub-tabs.
|
||||
// Wired to the same /workspaces/:id/a2a (method message/send) endpoint
|
||||
// that the desktop ChatTab uses. Render parity with desktop ChatTab is
|
||||
// achieved by reusing its renderers rather than forking a reduced
|
||||
// mobile path: the Agent Comms sub-tab mounts the same AgentCommsPanel,
|
||||
// and message attachments route through the same AttachmentPreview
|
||||
// dispatch the desktop My-Chat bubble uses (#231/#232).
|
||||
// that the desktop ChatTab uses, but with a slimmer surface: no
|
||||
// attachments, no A2A topology overlay, no conversation tracing.
|
||||
|
||||
import { useEffect, useMemo, useRef, useState } from "react";
|
||||
import ReactMarkdown from "react-markdown";
|
||||
@@ -19,9 +16,6 @@ import {
|
||||
useChatSend,
|
||||
useChatSocket,
|
||||
} from "@/components/tabs/chat/hooks";
|
||||
import { AgentCommsPanel } from "@/components/tabs/chat/AgentCommsPanel";
|
||||
import { AttachmentPreview } from "@/components/tabs/chat/AttachmentPreview";
|
||||
import { downloadChatFile } from "@/components/tabs/chat/uploads";
|
||||
|
||||
import { toMobileAgent } from "./components";
|
||||
import { MOBILE_FONT_MONO, MOBILE_FONT_SANS, usePalette } from "./palette";
|
||||
@@ -310,17 +304,6 @@ export function MobileChat({
|
||||
const removePendingFile = (index: number) =>
|
||||
setPendingFiles((prev) => prev.filter((_, i) => i !== index));
|
||||
|
||||
// Route attachment downloads through the same authenticated helper
|
||||
// the desktop ChatTab uses (downloadChatFile) so platform-scheme
|
||||
// URIs get a real Blob with auth headers instead of about:blank.
|
||||
const downloadAttachment = (att: ChatAttachment) => {
|
||||
downloadChatFile(agentId, att).catch(() => {
|
||||
// AttachmentPreview's own error affordance covers the in-bubble
|
||||
// failure state; matches ChatTab's behaviour of not double-
|
||||
// reporting a download failure.
|
||||
});
|
||||
};
|
||||
|
||||
const send = async () => {
|
||||
const text = draft.trim();
|
||||
if ((!text && pendingFiles.length === 0) || sending || !reachable) return;
|
||||
@@ -450,19 +433,7 @@ export function MobileChat({
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{/* Agent Comms — reuse the desktop AgentCommsPanel verbatim so
|
||||
mobile renders the identical peer/A2A + delegation feed
|
||||
(history GET + live socket events) instead of a placeholder
|
||||
(#231). The panel owns its own scroll/load/error/empty
|
||||
states, matching ChatTab's agent-comms tabpanel. */}
|
||||
{tab === "a2a" && (
|
||||
<div style={{ flex: 1, minHeight: 0, overflow: "hidden" }}>
|
||||
<AgentCommsPanel workspaceId={agentId} />
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* Messages */}
|
||||
{tab === "my" && (
|
||||
<div
|
||||
ref={scrollRef}
|
||||
style={{
|
||||
@@ -474,6 +445,18 @@ export function MobileChat({
|
||||
gap: 8,
|
||||
}}
|
||||
>
|
||||
{tab === "a2a" && (
|
||||
<div
|
||||
style={{
|
||||
padding: "20px 4px",
|
||||
textAlign: "center",
|
||||
color: p.text3,
|
||||
fontSize: 13,
|
||||
}}
|
||||
>
|
||||
Agent Comms — peer-to-peer A2A traffic surfaces in the Comms tab.
|
||||
</div>
|
||||
)}
|
||||
{tab === "my" && historyLoading && (
|
||||
<div style={{ padding: "20px 4px", textAlign: "center", color: p.text3, fontSize: 13 }}>
|
||||
Loading chat history…
|
||||
@@ -538,31 +521,9 @@ export function MobileChat({
|
||||
overflowWrap: "anywhere",
|
||||
}}
|
||||
>
|
||||
{m.content && (
|
||||
<MarkdownBubble dark={dark} accent={p.accent}>
|
||||
{m.content}
|
||||
</MarkdownBubble>
|
||||
)}
|
||||
{m.attachments && m.attachments.length > 0 && (
|
||||
<div
|
||||
style={{
|
||||
display: "flex",
|
||||
flexWrap: "wrap",
|
||||
gap: 4,
|
||||
marginTop: m.content ? 6 : 0,
|
||||
}}
|
||||
>
|
||||
{m.attachments.map((att, i) => (
|
||||
<AttachmentPreview
|
||||
key={`${m.id}-${i}`}
|
||||
workspaceId={agentId}
|
||||
attachment={att}
|
||||
onDownload={downloadAttachment}
|
||||
tone={mine ? "user" : "agent"}
|
||||
/>
|
||||
))}
|
||||
</div>
|
||||
)}
|
||||
<MarkdownBubble dark={dark} accent={p.accent}>
|
||||
{m.content}
|
||||
</MarkdownBubble>
|
||||
<div
|
||||
style={{
|
||||
fontSize: 10,
|
||||
@@ -593,13 +554,7 @@ export function MobileChat({
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* Footer ID + composer belong to My Chat only. The Agent Comms
|
||||
tab is a read-only peer/A2A feed (parity with desktop
|
||||
ChatTab, where the agent-comms tabpanel has no composer). */}
|
||||
{tab === "my" && (
|
||||
<>
|
||||
{/* Footer ID */}
|
||||
<div
|
||||
style={{
|
||||
@@ -791,8 +746,6 @@ export function MobileChat({
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
</>
|
||||
)}
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
@@ -21,14 +21,6 @@ import { MobileChat } from "../MobileChat";
|
||||
vi.mock("@/lib/api");
|
||||
import { api } from "@/lib/api";
|
||||
|
||||
// AgentCommsPanel (mounted by the Agent Comms sub-tab, #231) subscribes
|
||||
// to the global socket via useSocketEvent. Stub it to a no-op so the
|
||||
// panel mounts without the real ReconnectingSocket — the parity tests
|
||||
// only assert the panel renders (vs the old static placeholder).
|
||||
vi.mock("@/hooks/useSocketEvent", () => ({
|
||||
useSocketEvent: vi.fn(),
|
||||
}));
|
||||
|
||||
// ─── Mock store ───────────────────────────────────────────────────────────────
|
||||
|
||||
const mockAgentId = "ws-chat-test";
|
||||
@@ -163,12 +155,6 @@ beforeEach(() => {
|
||||
mockOnBack.mockClear();
|
||||
mockStoreState.nodes = [];
|
||||
mockStoreState.agentMessages = {};
|
||||
// jsdom doesn't implement scrollIntoView. The Agent Comms tab now
|
||||
// mounts AgentCommsPanel (#231), which scrolls its feed to bottom on
|
||||
// arrival; a no-op stub keeps the panel from throwing under jsdom
|
||||
// (same stub AgentCommsPanel's own render test installs).
|
||||
Element.prototype.scrollIntoView =
|
||||
vi.fn() as unknown as Element["scrollIntoView"];
|
||||
// Set up spies on the real api methods. Tests override these per-call.
|
||||
const getSpy = vi.spyOn(api, "get");
|
||||
const postSpy = vi.spyOn(api, "post");
|
||||
@@ -488,146 +474,3 @@ describe("MobileChat — chat history", () => {
|
||||
expect(getSpy).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
});
|
||||
|
||||
// ─── #232 · Attachment render parity with desktop ChatTab ────────────────────
|
||||
//
|
||||
// Regression for the CTO-reported mobile bug: MobileChat used to render
|
||||
// only m.content (no attachment surface), so files sent/received in a
|
||||
// conversation were invisible on mobile while desktop showed them. The
|
||||
// fix routes m.attachments through the same AttachmentPreview the
|
||||
// desktop ChatTab bubble uses.
|
||||
|
||||
describe("MobileChat — attachment render parity (#232)", () => {
|
||||
beforeEach(() => {
|
||||
mockStoreState.nodes = [onlineNode];
|
||||
});
|
||||
|
||||
it("renders an attachment from a history message via AttachmentPreview", async () => {
|
||||
const getSpy = vi.spyOn(api, "get");
|
||||
// useChatHistory reads { messages, reached_end }.
|
||||
getSpy.mockResolvedValueOnce({
|
||||
messages: [
|
||||
{
|
||||
id: "m-att-1",
|
||||
role: "agent",
|
||||
content: "Here is the report",
|
||||
attachments: [
|
||||
{
|
||||
name: "report.csv",
|
||||
uri: "workspace://out/report.csv",
|
||||
mimeType: "text/csv",
|
||||
size: 2048,
|
||||
},
|
||||
],
|
||||
timestamp: new Date().toISOString(),
|
||||
},
|
||||
],
|
||||
reached_end: true,
|
||||
});
|
||||
|
||||
let rr: ReturnType<typeof renderChat>;
|
||||
await act(async () => {
|
||||
rr = renderChat(mockAgentId);
|
||||
});
|
||||
const { container } = rr!;
|
||||
|
||||
// A non-image attachment renders the AttachmentChip download button
|
||||
// with title="Download <name>" — same component the desktop bubble
|
||||
// dispatches through AttachmentPreview.
|
||||
await waitFor(() => {
|
||||
const chip = container.querySelector('[title="Download report.csv"]');
|
||||
expect(chip).toBeTruthy();
|
||||
});
|
||||
expect(container.textContent ?? "").toContain("report.csv");
|
||||
});
|
||||
});
|
||||
|
||||
// ─── #231 · Agent Comms (A2A/peer) render parity with desktop ChatTab ────────
|
||||
//
|
||||
// Regression for the CTO-reported mobile bug: the Agent Comms sub-tab
|
||||
// rendered a static placeholder string ("peer-to-peer A2A traffic
|
||||
// surfaces in the Comms tab") instead of the real feed. The fix mounts
|
||||
// the same AgentCommsPanel the desktop ChatTab agent-comms tabpanel
|
||||
// uses, so peer/A2A + delegation activity is visible on mobile.
|
||||
|
||||
describe("MobileChat — Agent Comms render parity (#231)", () => {
|
||||
beforeEach(() => {
|
||||
mockStoreState.nodes = [onlineNode];
|
||||
});
|
||||
|
||||
it("mounts AgentCommsPanel on the Agent Comms tab (not the old placeholder)", async () => {
|
||||
const getSpy = vi.spyOn(api, "get");
|
||||
// 1st GET: useChatHistory (My Chat) on mount.
|
||||
getSpy.mockResolvedValueOnce({ messages: [], reached_end: true });
|
||||
// 2nd GET: AgentCommsPanel's activity load when the tab is shown.
|
||||
// Empty list → panel renders its own empty state, which still
|
||||
// proves AgentCommsPanel mounted (vs. the removed placeholder).
|
||||
getSpy.mockResolvedValueOnce([]);
|
||||
|
||||
let rr: ReturnType<typeof renderChat>;
|
||||
await act(async () => {
|
||||
rr = renderChat(mockAgentId);
|
||||
});
|
||||
const { container } = rr!;
|
||||
|
||||
const commsTab = Array.from(container.querySelectorAll("button")).find(
|
||||
(b) => b.textContent?.trim() === "Agent Comms",
|
||||
);
|
||||
expect(commsTab).toBeTruthy();
|
||||
await act(async () => {
|
||||
commsTab!.click();
|
||||
});
|
||||
|
||||
await waitFor(() => {
|
||||
const text = container.textContent ?? "";
|
||||
// The panel's empty state — proves AgentCommsPanel mounted.
|
||||
expect(text).toContain("No agent-to-agent communications yet.");
|
||||
});
|
||||
// The old hard-coded placeholder must be gone.
|
||||
expect(container.textContent ?? "").not.toContain(
|
||||
"peer-to-peer A2A traffic surfaces in the Comms tab",
|
||||
);
|
||||
// The panel hit its activity endpoint.
|
||||
expect(getSpy).toHaveBeenCalledWith(
|
||||
expect.stringContaining(`/workspaces/${mockAgentId}/activity`),
|
||||
);
|
||||
});
|
||||
|
||||
it("renders a peer message on the Agent Comms tab", async () => {
|
||||
const getSpy = vi.spyOn(api, "get");
|
||||
getSpy.mockResolvedValueOnce({ messages: [], reached_end: true });
|
||||
// a2a_receive from a peer → AgentCommsPanel.toCommMessage maps it
|
||||
// to an inbound bubble with the request text.
|
||||
getSpy.mockResolvedValueOnce([
|
||||
{
|
||||
id: "act-1",
|
||||
activity_type: "a2a_receive",
|
||||
source_id: "peer-ws-uuid",
|
||||
target_id: mockAgentId,
|
||||
method: "message/send",
|
||||
summary: "peer asked something",
|
||||
request_body: { task: "Please review PR 42" },
|
||||
response_body: null,
|
||||
status: "ok",
|
||||
created_at: new Date().toISOString(),
|
||||
},
|
||||
]);
|
||||
|
||||
let rr: ReturnType<typeof renderChat>;
|
||||
await act(async () => {
|
||||
rr = renderChat(mockAgentId);
|
||||
});
|
||||
const { container } = rr!;
|
||||
|
||||
const commsTab = Array.from(container.querySelectorAll("button")).find(
|
||||
(b) => b.textContent?.trim() === "Agent Comms",
|
||||
);
|
||||
await act(async () => {
|
||||
commsTab!.click();
|
||||
});
|
||||
|
||||
await waitFor(() => {
|
||||
expect(container.textContent ?? "").toContain("Please review PR 42");
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -399,7 +399,21 @@ func (h *WorkspaceHandler) proxyA2ARequest(ctx context.Context, workspaceID stri
|
||||
// (no Do(), no maybeMarkContainerDead). The response is a synthetic
|
||||
// {status:"queued"} envelope so the caller (canvas, another workspace)
|
||||
// knows delivery is acknowledged but pending consumption.
|
||||
if lookupDeliveryMode(ctx, workspaceID) == models.DeliveryModePoll {
|
||||
deliveryMode, deliveryModeErr := lookupDeliveryMode(ctx, workspaceID)
|
||||
if deliveryModeErr != nil {
|
||||
// internal#497 fail-closed: a real DB/context error on the
|
||||
// delivery-mode read MUST NOT silently fall through to the push
|
||||
// dispatch path — that is exactly what silently misrouted every
|
||||
// poll-mode peer for 5 days under the ce2db75f regression. Surface
|
||||
// a structured error so the delegation is marked failed (loud +
|
||||
// retryable) instead of dispatched to the wrong path.
|
||||
log.Printf("ProxyA2A: delivery-mode lookup failed for %s: %v — failing closed", workspaceID, deliveryModeErr)
|
||||
return 0, nil, &proxyA2AError{
|
||||
Status: http.StatusServiceUnavailable,
|
||||
Response: gin.H{"error": "delivery-mode lookup failed; refusing to dispatch to avoid silent misrouting"},
|
||||
}
|
||||
}
|
||||
if deliveryMode == models.DeliveryModePoll {
|
||||
if logActivity {
|
||||
h.logA2AReceiveQueued(ctx, workspaceID, callerID, body, a2aMethod)
|
||||
}
|
||||
|
||||
@@ -468,40 +468,64 @@ func parseUsageFromA2AResponse(body []byte) (inputTokens, outputTokens int64) {
|
||||
return 0, 0
|
||||
}
|
||||
|
||||
// lookupDeliveryMode returns the workspace's delivery_mode. On any DB
|
||||
// error or missing row it returns DeliveryModePush — the fail-closed
|
||||
// default. "Closed" here means "fall back to today's behavior (synchronous
|
||||
// dispatch)" rather than "fall back to drop the request silently into
|
||||
// activity_logs where the agent might never see it." A poll-mode workspace
|
||||
// that briefly reads as push will get its A2A request dispatched to the
|
||||
// stored URL (or a 502 if no URL); a push-mode workspace that briefly
|
||||
// reads as poll would get its request silently queued with no dispatch.
|
||||
// The first failure is loud + recoverable; the second is silent.
|
||||
// lookupDeliveryMode returns the workspace's delivery_mode.
|
||||
//
|
||||
// internal#497 / RFC#497 fail-closed (SURGICAL scope): the *specific*
|
||||
// failure mode that hid the ce2db75f regression for 5 days is now
|
||||
// propagated instead of silently swallowed — a CONTEXT error
|
||||
// (context.Canceled / context.DeadlineExceeded). Under ce2db75f the
|
||||
// detached delegation goroutine ran on a cancelled request context, every
|
||||
// `SELECT delivery_mode` failed `context canceled`, this function returned
|
||||
// push, the poll-mode short-circuit in proxyA2ARequest was skipped, and
|
||||
// poll-mode peers (e.g. an operator laptop on molecule-mcp-claude-channel)
|
||||
// silently never got their a2a_receive inbox row. A transient,
|
||||
// systematic-once-triggered context cancellation became permanent
|
||||
// invisible misrouting. Returning that error lets the caller fail loud
|
||||
// (mark the delegation failed) instead of mis-dispatching.
|
||||
//
|
||||
// Scope is deliberately narrow: only ctx errors propagate. Other DB
|
||||
// errors retain the long-standing documented "fall back to push (today's
|
||||
// synchronous behavior)" contract — that path is loud + recoverable
|
||||
// (502 / SSRF reject / restart), unlike the silent poll-mode drop, and
|
||||
// the surrounding proxy (incl. the sibling checkWorkspaceBudget) is
|
||||
// intentionally built around that fail-open-to-push behavior. Widening
|
||||
// further is an RFC#497 follow-up, not part of this P0 fix.
|
||||
//
|
||||
// A genuinely *absent* configuration is NOT an error and still resolves to
|
||||
// push (the safe synchronous default): sql.ErrNoRows, a NULL/empty column,
|
||||
// or an unrecognised value all return (push, nil).
|
||||
//
|
||||
// The function is intentionally lookup-only — it never mutates the row.
|
||||
// The register handler (registry.go) is the only writer for delivery_mode.
|
||||
//
|
||||
// See #2339 PR 1 for the column + register-flow side; this is the
|
||||
// proxy-side read used for the short-circuit in proxyA2ARequest.
|
||||
func lookupDeliveryMode(ctx context.Context, workspaceID string) string {
|
||||
func lookupDeliveryMode(ctx context.Context, workspaceID string) (string, error) {
|
||||
var mode sql.NullString
|
||||
err := db.DB.QueryRowContext(ctx,
|
||||
`SELECT delivery_mode FROM workspaces WHERE id = $1`, workspaceID,
|
||||
).Scan(&mode)
|
||||
if err != nil {
|
||||
if !errors.Is(err, sql.ErrNoRows) {
|
||||
log.Printf("ProxyA2A: lookupDeliveryMode(%s) failed (%v) — defaulting to push", workspaceID, err)
|
||||
// internal#497: a context cancellation/deadline MUST NOT be
|
||||
// swallowed into a silent push default — that is the exact 5-day
|
||||
// silent-misrouting vector. Propagate so the caller fails closed.
|
||||
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
|
||||
log.Printf("ProxyA2A: lookupDeliveryMode(%s) context error (%v) — failing closed (NOT defaulting to push)", workspaceID, err)
|
||||
return "", err
|
||||
}
|
||||
return models.DeliveryModePush
|
||||
if !errors.Is(err, sql.ErrNoRows) {
|
||||
log.Printf("ProxyA2A: lookupDeliveryMode(%s) failed (%v) — defaulting to push (non-ctx DB error; legacy fail-open-to-push contract)", workspaceID, err)
|
||||
}
|
||||
return models.DeliveryModePush, nil
|
||||
}
|
||||
if !mode.Valid || mode.String == "" {
|
||||
return models.DeliveryModePush
|
||||
return models.DeliveryModePush, nil
|
||||
}
|
||||
if !models.IsValidDeliveryMode(mode.String) {
|
||||
log.Printf("ProxyA2A: workspace %s has invalid delivery_mode=%q — defaulting to push", workspaceID, mode.String)
|
||||
return models.DeliveryModePush
|
||||
return models.DeliveryModePush, nil
|
||||
}
|
||||
return mode.String
|
||||
return mode.String, nil
|
||||
}
|
||||
|
||||
// logA2AReceiveQueued records a poll-mode "queued" A2A receive into
|
||||
|
||||
@@ -2228,12 +2228,18 @@ func TestProxyA2A_PushMode_NoShortCircuit(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestProxyA2A_PollMode_FailsClosedToPush verifies the safety contract:
|
||||
// a DB error reading delivery_mode must default to push (the existing
|
||||
// behavior), NOT poll. Failing to push means a poll-mode workspace
|
||||
// briefly attempts a real dispatch — visible failure (502 / SSRF
|
||||
// rejection / restart cascade), not a silent drop into activity_logs
|
||||
// where the agent might never look. Loud > silent, recoverable > lost.
|
||||
// TestProxyA2A_PollMode_FailsClosedToPush verifies the LEGACY safety
|
||||
// contract is PRESERVED for non-context DB errors: a generic DB error
|
||||
// reading delivery_mode still defaults to push (today's behavior), NOT
|
||||
// poll. Failing to push means a poll-mode workspace briefly attempts a
|
||||
// real dispatch — visible failure (502 / SSRF rejection / restart
|
||||
// cascade), not a silent drop into activity_logs where the agent might
|
||||
// never look. Loud > silent, recoverable > lost.
|
||||
//
|
||||
// internal#497 narrows the fail-closed change to *context* errors only
|
||||
// (the actual ce2db75f regression vector); generic DB errors keep this
|
||||
// long-standing fail-open-to-push contract. The ctx-error fail-closed is
|
||||
// covered by TestLookupDeliveryMode_ContextCanceled_FailsClosed.
|
||||
func TestProxyA2A_PollMode_FailsClosedToPush(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
setupTestRedis(t) // empty Redis — forces resolveAgentURL DB lookup
|
||||
@@ -2244,7 +2250,8 @@ func TestProxyA2A_PollMode_FailsClosedToPush(t *testing.T) {
|
||||
|
||||
expectBudgetCheck(mock, wsID)
|
||||
|
||||
// lookupDeliveryMode hits a transient DB error → must default push.
|
||||
// lookupDeliveryMode hits a generic (non-context) DB error → must
|
||||
// still default push (legacy contract preserved by internal#497).
|
||||
mock.ExpectQuery("SELECT delivery_mode FROM workspaces WHERE id").
|
||||
WithArgs(wsID).
|
||||
WillReturnError(sql.ErrConnDone)
|
||||
@@ -2268,7 +2275,7 @@ func TestProxyA2A_PollMode_FailsClosedToPush(t *testing.T) {
|
||||
var resp map[string]interface{}
|
||||
_ = json.Unmarshal(w.Body.Bytes(), &resp)
|
||||
if resp["status"] == "queued" {
|
||||
t.Errorf("DB error on delivery_mode lookup silently queued the request — must fail-closed-to-push, got body: %s", w.Body.String())
|
||||
t.Errorf("generic DB error on delivery_mode lookup silently queued the request — must fail-open-to-push, got body: %s", w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2277,6 +2284,37 @@ func TestProxyA2A_PollMode_FailsClosedToPush(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestLookupDeliveryMode_ContextCanceled_FailsClosed is the internal#497
|
||||
// regression test for the SECONDARY defect. It pins the exact invariant
|
||||
// that hid the ce2db75f regression for 5 days: when the delivery_mode read
|
||||
// fails because the context was cancelled (precisely what happened in the
|
||||
// detached delegation goroutine running on a returned request context),
|
||||
// lookupDeliveryMode MUST return an error and MUST NOT silently return
|
||||
// "push". Returning push there is what skipped the poll-mode short-circuit
|
||||
// and silently dropped 100% of poll-mode peer deliveries.
|
||||
//
|
||||
// A pre-cancelled context makes QueryRowContext fail with
|
||||
// context.Canceled deterministically — no DB rows are mocked because the
|
||||
// query never reaches a result.
|
||||
func TestLookupDeliveryMode_ContextCanceled_FailsClosed(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
// The query fails on the cancelled ctx before matching; provide a
|
||||
// permissive expectation so sqlmock doesn't complain about the attempt.
|
||||
mock.ExpectQuery("SELECT delivery_mode FROM workspaces WHERE id").
|
||||
WillReturnError(context.Canceled)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel() // simulate the HTTP handler having returned (request ctx dead)
|
||||
|
||||
mode, err := lookupDeliveryMode(ctx, "ws-poll-peer")
|
||||
if err == nil {
|
||||
t.Fatalf("internal#497 regression: lookupDeliveryMode swallowed a context error and returned mode=%q with nil err — this is the exact 5-day silent-misrouting vector", mode)
|
||||
}
|
||||
if mode == models.DeliveryModePush {
|
||||
t.Errorf("internal#497 regression: context error must NOT default to push (got mode=%q)", mode)
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== a2aClient ResponseHeaderTimeout config ====================
|
||||
|
||||
func TestA2AClientResponseHeaderTimeout(t *testing.T) {
|
||||
|
||||
@@ -2,224 +2,206 @@ package handlers
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/DATA-DOG/go-sqlmock"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/db"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/wsauth"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
func newTestTokenRequest(workspaceID string) (*httptest.ResponseRecorder, *gin.Context) {
|
||||
// Valid UUID used throughout.
|
||||
const wsToken = "00000000-0000-0000-0000-000000000030"
|
||||
|
||||
// ---------- TestTokensEnabled ----------
|
||||
|
||||
func TestTokensEnabled_EnvFlagTrue(t *testing.T) {
|
||||
t.Setenv("MOLECULE_ENABLE_TEST_TOKENS", "1")
|
||||
t.Setenv("MOLECULE_ENV", "production")
|
||||
if !TestTokensEnabled() {
|
||||
t.Error("expected true when MOLECULE_ENABLE_TEST_TOKENS=1")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTokensEnabled_ProductionEnv(t *testing.T) {
|
||||
t.Setenv("MOLECULE_ENABLE_TEST_TOKENS", "")
|
||||
t.Setenv("MOLECULE_ENV", "production")
|
||||
if TestTokensEnabled() {
|
||||
t.Error("expected false when MOLECULE_ENV=production")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTokensEnabled_StagingEnv(t *testing.T) {
|
||||
t.Setenv("MOLECULE_ENABLE_TEST_TOKENS", "")
|
||||
t.Setenv("MOLECULE_ENV", "staging")
|
||||
if !TestTokensEnabled() {
|
||||
t.Error("expected true when MOLECULE_ENV=staging")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTokensEnabled_EmptyEnv(t *testing.T) {
|
||||
t.Setenv("MOLECULE_ENABLE_TEST_TOKENS", "")
|
||||
t.Setenv("MOLECULE_ENV", "")
|
||||
if !TestTokensEnabled() {
|
||||
t.Error("expected true when MOLECULE_ENV is empty (local dev default)")
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- GetTestToken ----------
|
||||
|
||||
func makeTokenHandler(t *testing.T) (*AdminTestTokenHandler, sqlmock.Sqlmock, func()) {
|
||||
t.Helper()
|
||||
mockDB, mock, err := sqlmock.New()
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create sqlmock: %v", err)
|
||||
}
|
||||
prevDB := db.DB
|
||||
db.DB = mockDB
|
||||
return NewAdminTestTokenHandler(), mock, func() {
|
||||
db.DB = prevDB
|
||||
mockDB.Close()
|
||||
}
|
||||
}
|
||||
|
||||
func getTestToken(t *testing.T, h *AdminTestTokenHandler, workspaceID string, adminToken string) *httptest.ResponseRecorder {
|
||||
t.Helper()
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Params = gin.Params{{Key: "id", Value: workspaceID}}
|
||||
c.Request = httptest.NewRequest("GET", "/admin/workspaces/"+workspaceID+"/test-token", nil)
|
||||
return w, c
|
||||
req := httptest.NewRequest("GET", "/admin/workspaces/"+workspaceID+"/test-token", nil)
|
||||
if adminToken != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+adminToken)
|
||||
}
|
||||
c.Request = req
|
||||
h.GetTestToken(c)
|
||||
return w
|
||||
}
|
||||
|
||||
func TestAdminTestToken_HiddenInProduction(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
t.Setenv("MOLECULE_ENV", "production")
|
||||
func TestGetTestToken_DisabledByDefault(t *testing.T) {
|
||||
// Set MOLECULE_ENV=production to simulate a locked-down environment.
|
||||
t.Setenv("MOLECULE_ENABLE_TEST_TOKENS", "")
|
||||
|
||||
h := NewAdminTestTokenHandler()
|
||||
w, c := newTestTokenRequest("ws-1")
|
||||
h.GetTestToken(c)
|
||||
|
||||
if w.Code != http.StatusNotFound {
|
||||
t.Fatalf("expected 404 in production, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminTestToken_EnabledViaFlagEvenInProd(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
t.Setenv("MOLECULE_ENV", "production")
|
||||
t.Setenv("MOLECULE_ENABLE_TEST_TOKENS", "1")
|
||||
|
||||
mock.ExpectQuery("SELECT id FROM workspaces WHERE id =").
|
||||
WithArgs("ws-1").
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow("ws-1"))
|
||||
mock.ExpectExec("INSERT INTO workspace_auth_tokens").
|
||||
WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
|
||||
h := NewAdminTestTokenHandler()
|
||||
w, c := newTestTokenRequest("ws-1")
|
||||
h.GetTestToken(c)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminTestToken_WorkspaceNotFound(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
t.Setenv("MOLECULE_ENV", "development")
|
||||
|
||||
mock.ExpectQuery("SELECT id FROM workspaces WHERE id =").
|
||||
WithArgs("missing").
|
||||
WillReturnError(sqlErrNoRows())
|
||||
|
||||
h := NewAdminTestTokenHandler()
|
||||
w, c := newTestTokenRequest("missing")
|
||||
h.GetTestToken(c)
|
||||
|
||||
w := getTestToken(t, h, wsToken, "")
|
||||
if w.Code != http.StatusNotFound {
|
||||
t.Fatalf("expected 404 for missing workspace, got %d: %s", w.Code, w.Body.String())
|
||||
t.Errorf("expected 404 when disabled, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminTestToken_HappyPath_TokenValidates(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
t.Setenv("MOLECULE_ENV", "development")
|
||||
|
||||
mock.ExpectQuery("SELECT id FROM workspaces WHERE id =").
|
||||
WithArgs("ws-1").
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow("ws-1"))
|
||||
|
||||
// Capture the hash inserted by IssueToken so we can replay it on Validate.
|
||||
var capturedHash []byte
|
||||
mock.ExpectExec("INSERT INTO workspace_auth_tokens").
|
||||
WithArgs("ws-1", sqlmock.AnyArg(), sqlmock.AnyArg()).
|
||||
WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
func TestGetTestToken_AdminTokenRequired_WrongToken(t *testing.T) {
|
||||
// Set up: tokens enabled, ADMIN_TOKEN set, but request uses wrong token.
|
||||
t.Setenv("MOLECULE_ENABLE_TEST_TOKENS", "1")
|
||||
t.Setenv("MOLECULE_ENV", "production")
|
||||
os.Setenv("ADMIN_TOKEN", "correct-secret")
|
||||
defer os.Unsetenv("ADMIN_TOKEN")
|
||||
|
||||
h := NewAdminTestTokenHandler()
|
||||
w, c := newTestTokenRequest("ws-1")
|
||||
h.GetTestToken(c)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
|
||||
var resp struct {
|
||||
AuthToken string `json:"auth_token"`
|
||||
WorkspaceID string `json:"workspace_id"`
|
||||
}
|
||||
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
|
||||
t.Fatalf("bad json: %v", err)
|
||||
}
|
||||
if resp.AuthToken == "" {
|
||||
t.Fatal("expected non-empty auth_token")
|
||||
}
|
||||
if resp.WorkspaceID != "ws-1" {
|
||||
t.Errorf("expected workspace_id ws-1, got %q", resp.WorkspaceID)
|
||||
}
|
||||
if len(resp.AuthToken) < 32 {
|
||||
t.Errorf("token looks too short: %d chars", len(resp.AuthToken))
|
||||
}
|
||||
|
||||
// Now simulate ValidateToken lookup using the same DB — prove the token
|
||||
// can be validated by feeding its sha256 back through ExpectedArgs.
|
||||
// (We stub the SELECT rather than re-reading capturedHash since sqlmock
|
||||
// doesn't capture live args; the important invariant is that the issued
|
||||
// token passes ValidateToken given a matching hash row exists.)
|
||||
_ = capturedHash
|
||||
mock.ExpectQuery("SELECT t\\.id, t\\.workspace_id.*FROM workspace_auth_tokens t.*JOIN workspaces").
|
||||
WithArgs(sqlmock.AnyArg()).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id", "workspace_id"}).AddRow("tok-1", "ws-1"))
|
||||
mock.ExpectExec("UPDATE workspace_auth_tokens SET last_used_at").
|
||||
WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
|
||||
if err := wsauth.ValidateToken(c.Request.Context(), db.DB, "ws-1", resp.AuthToken); err != nil {
|
||||
t.Errorf("issued token failed to validate: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func sqlErrNoRows() error { return sql.ErrNoRows }
|
||||
|
||||
// TestAdminTestToken_AdminTokenRequired_NoHeader pins the IDOR-fix (#112):
|
||||
// when ADMIN_TOKEN is set, calls without an Authorization header MUST 401.
|
||||
// Pre-fix, the route accepted any bearer that matched a live org token,
|
||||
// allowing cross-org test-token minting. The current code uses
|
||||
// subtle.ConstantTimeCompare against ADMIN_TOKEN explicitly. This test
|
||||
// pins that no-header == 401 so a regression that re-enabled the AdminAuth
|
||||
// fallback would fail loudly.
|
||||
func TestAdminTestToken_AdminTokenRequired_NoHeader(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
t.Setenv("MOLECULE_ENV", "development")
|
||||
t.Setenv("ADMIN_TOKEN", "the-admin-secret")
|
||||
|
||||
h := NewAdminTestTokenHandler()
|
||||
w, c := newTestTokenRequest("ws-1")
|
||||
h.GetTestToken(c)
|
||||
|
||||
w := getTestToken(t, h, wsToken, "wrong-token")
|
||||
if w.Code != http.StatusUnauthorized {
|
||||
t.Fatalf("expected 401 with ADMIN_TOKEN set + no Authorization, got %d: %s", w.Code, w.Body.String())
|
||||
t.Errorf("expected 401, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdminTestToken_AdminTokenRequired_WrongHeader pins that a non-matching
|
||||
// bearer is rejected. Critical for #112 — an attacker presenting any other
|
||||
// org's token must NOT pass.
|
||||
func TestAdminTestToken_AdminTokenRequired_WrongHeader(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
t.Setenv("MOLECULE_ENV", "development")
|
||||
t.Setenv("ADMIN_TOKEN", "the-admin-secret")
|
||||
func TestGetTestToken_AdminTokenRequired_MissingBearer(t *testing.T) {
|
||||
t.Setenv("MOLECULE_ENABLE_TEST_TOKENS", "1")
|
||||
t.Setenv("MOLECULE_ENV", "production")
|
||||
os.Setenv("ADMIN_TOKEN", "correct-secret")
|
||||
defer os.Unsetenv("ADMIN_TOKEN")
|
||||
|
||||
h := NewAdminTestTokenHandler()
|
||||
w, c := newTestTokenRequest("ws-1")
|
||||
c.Request.Header.Set("Authorization", "Bearer wrong-token")
|
||||
h.GetTestToken(c)
|
||||
|
||||
w := getTestToken(t, h, wsToken, "")
|
||||
if w.Code != http.StatusUnauthorized {
|
||||
t.Fatalf("expected 401 with wrong Authorization, got %d: %s", w.Code, w.Body.String())
|
||||
t.Errorf("expected 401 when bearer missing, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdminTestToken_AdminTokenRequired_CorrectHeader pins the success
|
||||
// path through the ADMIN_TOKEN gate. Together with the no-header + wrong-
|
||||
// header pair, this proves the gate distinguishes correct from incorrect
|
||||
// rather than (e.g.) erroring on every request.
|
||||
func TestAdminTestToken_AdminTokenRequired_CorrectHeader(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
t.Setenv("MOLECULE_ENV", "development")
|
||||
t.Setenv("ADMIN_TOKEN", "the-admin-secret")
|
||||
func TestGetTestToken_AdminTokenRequired_CorrectToken(t *testing.T) {
|
||||
t.Setenv("MOLECULE_ENABLE_TEST_TOKENS", "1")
|
||||
t.Setenv("MOLECULE_ENV", "production")
|
||||
os.Setenv("ADMIN_TOKEN", "correct-secret")
|
||||
defer os.Unsetenv("ADMIN_TOKEN")
|
||||
|
||||
mock.ExpectQuery("SELECT id FROM workspaces WHERE id =").
|
||||
WithArgs("ws-1").
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow("ws-1"))
|
||||
mock.ExpectExec("INSERT INTO workspace_auth_tokens").
|
||||
_, mock, cleanup := makeTokenHandler(t)
|
||||
defer cleanup()
|
||||
|
||||
mock.ExpectQuery(`SELECT id FROM workspaces WHERE id = \$1`).
|
||||
WithArgs(wsToken).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(wsToken))
|
||||
// IssueToken returns a token — we just need to verify the query ran.
|
||||
mock.ExpectExec(`INSERT INTO workspace_auth_tokens`).
|
||||
WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
|
||||
h := NewAdminTestTokenHandler()
|
||||
w, c := newTestTokenRequest("ws-1")
|
||||
c.Request.Header.Set("Authorization", "Bearer the-admin-secret")
|
||||
h.GetTestToken(c)
|
||||
|
||||
w := getTestToken(t, h, wsToken, "correct-secret")
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("expected 200 with correct ADMIN_TOKEN, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Errorf("sqlmock expectations not met — INSERT into workspace_auth_tokens did not run, suggesting the gate short-circuited the success path: %v", err)
|
||||
t.Errorf("expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdminTestToken_AdminTokenEmpty_GateBypassedSafely pins that when
|
||||
// ADMIN_TOKEN is unset (typical local-dev setup), the explicit gate is
|
||||
// bypassed and the route works without an Authorization header. This is
|
||||
// the same code path the existing TestAdminTestToken_EnabledViaFlagEvenInProd
|
||||
// exercises, but pinned explicitly so a future refactor that conflates
|
||||
// "ADMIN_TOKEN unset" with "always 401" gets caught immediately.
|
||||
func TestAdminTestToken_AdminTokenEmpty_GateBypassedSafely(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
t.Setenv("MOLECULE_ENV", "development")
|
||||
t.Setenv("ADMIN_TOKEN", "")
|
||||
func TestGetTestToken_WorkspaceNotFound(t *testing.T) {
|
||||
t.Setenv("MOLECULE_ENABLE_TEST_TOKENS", "1")
|
||||
t.Setenv("MOLECULE_ENV", "production")
|
||||
// ADMIN_TOKEN not set — no auth header required.
|
||||
|
||||
mock.ExpectQuery("SELECT id FROM workspaces WHERE id =").
|
||||
WithArgs("ws-1").
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow("ws-1"))
|
||||
mock.ExpectExec("INSERT INTO workspace_auth_tokens").
|
||||
_, mock, cleanup := makeTokenHandler(t)
|
||||
defer cleanup()
|
||||
|
||||
mock.ExpectQuery(`SELECT id FROM workspaces WHERE id = \$1`).
|
||||
WithArgs(wsToken).
|
||||
WillReturnError(sql.ErrNoRows)
|
||||
|
||||
h := NewAdminTestTokenHandler()
|
||||
w := getTestToken(t, h, wsToken, "")
|
||||
if w.Code != http.StatusNotFound {
|
||||
t.Errorf("expected 404 for missing workspace, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetTestToken_IssueTokenDBError(t *testing.T) {
|
||||
t.Setenv("MOLECULE_ENABLE_TEST_TOKENS", "1")
|
||||
t.Setenv("MOLECULE_ENV", "production")
|
||||
|
||||
_, mock, cleanup := makeTokenHandler(t)
|
||||
defer cleanup()
|
||||
|
||||
mock.ExpectQuery(`SELECT id FROM workspaces WHERE id = \$1`).
|
||||
WithArgs(wsToken).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(wsToken))
|
||||
// IssueToken fails.
|
||||
mock.ExpectExec(`INSERT INTO workspace_auth_tokens`).
|
||||
WillReturnError(sql.ErrConnDone)
|
||||
|
||||
h := NewAdminTestTokenHandler()
|
||||
w := getTestToken(t, h, wsToken, "")
|
||||
if w.Code != http.StatusInternalServerError {
|
||||
t.Errorf("expected 500 on token issue failure, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetTestToken_ResponseContainsToken(t *testing.T) {
|
||||
t.Setenv("MOLECULE_ENABLE_TEST_TOKENS", "1")
|
||||
t.Setenv("MOLECULE_ENV", "production")
|
||||
|
||||
_, mock, cleanup := makeTokenHandler(t)
|
||||
defer cleanup()
|
||||
|
||||
mock.ExpectQuery(`SELECT id FROM workspaces WHERE id = \$1`).
|
||||
WithArgs(wsToken).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(wsToken))
|
||||
mock.ExpectExec(`INSERT INTO workspace_auth_tokens`).
|
||||
WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
|
||||
h := NewAdminTestTokenHandler()
|
||||
w, c := newTestTokenRequest("ws-1")
|
||||
// Note: NO Authorization header — the gate is unset, so this MUST work.
|
||||
h.GetTestToken(c)
|
||||
|
||||
w := getTestToken(t, h, wsToken, "")
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("expected 200 with ADMIN_TOKEN empty + no Authorization, got %d: %s", w.Code, w.Body.String())
|
||||
t.Errorf("expected 200, got %d", w.Code)
|
||||
}
|
||||
body := w.Body.String()
|
||||
if !(strings.Contains(body, "auth_token") && strings.Contains(body, wsToken)) {
|
||||
t.Errorf("expected auth_token in response body, got: %s", body)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -162,8 +162,32 @@ func (h *DelegationHandler) Delegate(c *gin.Context) {
|
||||
},
|
||||
})
|
||||
|
||||
// Fire-and-forget: send A2A in background goroutine
|
||||
go h.executeDelegation(ctx, sourceID, body.TargetID, delegationID, a2aBody)
|
||||
// Fire-and-forget: send A2A in a background goroutine.
|
||||
//
|
||||
// internal#497 — the goroutine MUST NOT inherit the HTTP request's
|
||||
// cancellation. `ctx` here is c.Request.Context(); the handler returns
|
||||
// 202 a few lines below, which cancels that context immediately. Before
|
||||
// this fix (regression ce2db75f) executeDelegation ran on the
|
||||
// request-scoped ctx, so every DB op + proxy call in the detached
|
||||
// goroutine failed `context canceled` the instant the 202 was written.
|
||||
// That silently broke 100% of A2A peer delegations fleet-wide since
|
||||
// 2026-05-12 (poll-mode peers never got their a2a_receive inbox row;
|
||||
// lookupDeliveryMode swallowed the ctx error and defaulted to push).
|
||||
//
|
||||
// context.WithoutCancel detaches cancellation/deadline while PRESERVING
|
||||
// all context values (trace/correlation/tenant ids that proxyA2ARequest
|
||||
// and the broadcaster read off ctx) — this is the established pattern in
|
||||
// this package (a2a_proxy.go:850, a2a_proxy_helpers.go:525,
|
||||
// registry.go:822). The 30-minute ceiling matches the prior internal
|
||||
// budget executeDelegation used before ce2db75f and the proxy's own
|
||||
// absolute agent-dispatch ceiling (a2a_proxy.go forwardCtx).
|
||||
delegationCtx, cancelDelegation := context.WithTimeout(
|
||||
context.WithoutCancel(ctx), 30*time.Minute,
|
||||
)
|
||||
go func() {
|
||||
defer cancelDelegation()
|
||||
h.executeDelegation(delegationCtx, sourceID, body.TargetID, delegationID, a2aBody)
|
||||
}()
|
||||
|
||||
// Broadcast event so canvas shows delegation in real-time
|
||||
h.broadcaster.RecordAndBroadcast(ctx, string(events.EventDelegationSent), sourceID, map[string]interface{}{
|
||||
|
||||
@@ -16,6 +16,65 @@ import (
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// ---------- internal#497 regression: detached goroutine ctx must outlive the handler ----------
|
||||
|
||||
// TestDelegate_DetachedContext_SurvivesRequestCancellation pins the
|
||||
// load-bearing invariant that regression ce2db75f violated: the context
|
||||
// handed to executeDelegation in the fire-and-forget goroutine must NOT be
|
||||
// cancelled when the HTTP handler returns 202 (which cancels
|
||||
// c.Request.Context()). Before the fix, executeDelegation ran on the
|
||||
// request-scoped ctx, so every DB op + proxy call failed `context
|
||||
// canceled` the instant the 202 was written — silently breaking 100% of
|
||||
// A2A peer delegations fleet-wide since 2026-05-12.
|
||||
//
|
||||
// This test asserts the exact ctx-derivation contract used by Delegate
|
||||
// (context.WithoutCancel(parent) + a timeout budget): the derived context
|
||||
// (a) stays alive after the parent is cancelled, and (b) still carries
|
||||
// parent values (trace/correlation/tenant ids the downstream proxy +
|
||||
// broadcaster read off ctx). It is intentionally DB-free and fast.
|
||||
func TestDelegate_DetachedContext_SurvivesRequestCancellation(t *testing.T) {
|
||||
type ctxKey string
|
||||
const traceKey ctxKey = "trace-id"
|
||||
|
||||
// Simulate c.Request.Context() carrying a correlation value.
|
||||
parent, cancelParent := context.WithCancel(
|
||||
context.WithValue(context.Background(), traceKey, "trace-abc-123"),
|
||||
)
|
||||
|
||||
// Exact derivation Delegate uses for the detached goroutine.
|
||||
delegationCtx, cancelDelegation := context.WithTimeout(
|
||||
context.WithoutCancel(parent), 30*time.Minute,
|
||||
)
|
||||
defer cancelDelegation()
|
||||
|
||||
// The HTTP handler "returns 202" → request context is cancelled.
|
||||
cancelParent()
|
||||
|
||||
if err := parent.Err(); err == nil {
|
||||
t.Fatal("precondition: parent context should be cancelled after the handler returns")
|
||||
}
|
||||
|
||||
// (a) Cancellation MUST NOT propagate to the detached context.
|
||||
select {
|
||||
case <-delegationCtx.Done():
|
||||
t.Fatalf("regression: detached delegation ctx was cancelled by the handler returning (err=%v) — executeDelegation would fail every DB op with `context canceled`", delegationCtx.Err())
|
||||
default:
|
||||
// alive — correct
|
||||
}
|
||||
|
||||
// (b) Parent values MUST still be readable (WithoutCancel preserves
|
||||
// values; trace/correlation/tenant ids the proxy + broadcaster use).
|
||||
if got, _ := delegationCtx.Value(traceKey).(string); got != "trace-abc-123" {
|
||||
t.Errorf("detached ctx lost the parent trace value: got %q, want %q", got, "trace-abc-123")
|
||||
}
|
||||
|
||||
// And it still has a real deadline (the 30m budget), so it is not an
|
||||
// unbounded background context.
|
||||
if _, hasDeadline := delegationCtx.Deadline(); !hasDeadline {
|
||||
t.Error("detached ctx must carry the 30-minute timeout budget, but has no deadline")
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- Delegate: missing target_id → 400 ----------
|
||||
|
||||
func TestDelegate_MissingTargetID(t *testing.T) {
|
||||
|
||||
Reference in New Issue
Block a user