diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 826d96a..deb6715 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -153,6 +153,10 @@ description and keep ownership on the side listed here. conversation has received no new prompt for one hour. A new prompt for the same `AgentStateKey` renews that idle window. Explicit cancellation, device shutdown, and daemon shutdown still terminate processes immediately. +- Run terminal-state persistence must use a short independent context. A + dispatch deadline or cancellation may stop connector work, but it must not + prevent the server from recording the resulting completed, failed, or + cancelled state. - When an engine supports resume, persist the upstream session id through `agent_engine_sessions` and pass `AgentSessionID` plus `AgentStateKey` over the daemon protocol. Do not keep resume ids only in adapter memory, files diff --git a/apps/parsar-daemon/internal/agent/claudecode/options.go b/apps/parsar-daemon/internal/agent/claudecode/options.go index 147da43..e8679a5 100644 --- a/apps/parsar-daemon/internal/agent/claudecode/options.go +++ b/apps/parsar-daemon/internal/agent/claudecode/options.go @@ -32,6 +32,7 @@ func BuildArgs(opts map[string]any, resumeSessionID string) (BuildResult, error) args := []string{ "--output-format", "stream-json", "--input-format", "stream-json", + "--include-partial-messages", "--verbose", "--permission-prompt-tool", "stdio", } diff --git a/apps/parsar-daemon/internal/agent/claudecode/options_test.go b/apps/parsar-daemon/internal/agent/claudecode/options_test.go index e236c5e..4140b1a 100644 --- a/apps/parsar-daemon/internal/agent/claudecode/options_test.go +++ b/apps/parsar-daemon/internal/agent/claudecode/options_test.go @@ -31,6 +31,9 @@ func TestBuildArgsBaseHasStreamFlags(t *testing.T) { if !slices.Contains(res.Args, "--verbose") { t.Errorf("missing --verbose in %v", res.Args) } + if !slices.Contains(res.Args, "--include-partial-messages") { + t.Errorf("missing --include-partial-messages in %v", res.Args) + } } // IS_SANDBOX=1 must be in every env passthrough. Without it Claude diff --git a/apps/parsar-daemon/internal/agent/claudecode/parser.go b/apps/parsar-daemon/internal/agent/claudecode/parser.go index 883e8ad..998e563 100644 --- a/apps/parsar-daemon/internal/agent/claudecode/parser.go +++ b/apps/parsar-daemon/internal/agent/claudecode/parser.go @@ -58,12 +58,13 @@ func defaultAskIDMinter() string { // translator converts one NDJSON line from claude stdout into zero or // more proto.Envelope frames. One translator lives per session. type translator struct { - runID string - pending pendingRecorder - askPending askRecorder - seq atomic.Uint64 - mint permIDMinter - askMint askIDMinter + runID string + pending pendingRecorder + askPending askRecorder + seq atomic.Uint64 + mint permIDMinter + askMint askIDMinter + partialBlocks map[int]string } func newTranslator(runID string, pending pendingRecorder, askPending askRecorder, mint permIDMinter, askMint askIDMinter) *translator { @@ -116,6 +117,8 @@ func (t *translator) Translate(line []byte) (translation, error) { return t.translateSystem(line) case "assistant": return t.translateAssistant(line) + case "stream_event": + return t.translateStreamEvent(line) case "user": return t.translateUser(line) case "control_request": @@ -129,6 +132,67 @@ func (t *translator) Translate(line []byte) (translation, error) { } } +// translateStreamEvent handles the raw Anthropic events emitted by Claude +// Code with --include-partial-messages. Claude still emits a complete +// assistant frame after these events, so translateAssistant suppresses its +// text/thinking copies once a corresponding partial delta has been observed. +func (t *translator) translateStreamEvent(line []byte) (translation, error) { + var msg struct { + Event struct { + Type string `json:"type"` + Index int `json:"index"` + Delta struct { + Type string `json:"type"` + Text string `json:"text"` + Thinking string `json:"thinking"` + } `json:"delta"` + } `json:"event"` + } + if err := json.Unmarshal(line, &msg); err != nil { + return translation{}, fmt.Errorf("claudecode: parse stream_event frame: %w", err) + } + if msg.Event.Type != "content_block_delta" { + return translation{}, nil + } + + switch msg.Event.Delta.Type { + case "text_delta": + if msg.Event.Delta.Text == "" { + return translation{}, nil + } + if t.partialBlocks == nil { + t.partialBlocks = make(map[int]string) + } + t.partialBlocks[msg.Event.Index] = "text" + env, err := proto.NewEnvelope(proto.TypeDelta, t.runID, proto.DeltaPayload{ + Delta: msg.Event.Delta.Text, + Sequence: t.seq.Add(1), + }) + if err != nil { + return translation{}, err + } + return translation{Envelopes: []proto.Envelope{env}}, nil + case "thinking_delta": + if msg.Event.Delta.Thinking == "" { + return translation{}, nil + } + if t.partialBlocks == nil { + t.partialBlocks = make(map[int]string) + } + t.partialBlocks[msg.Event.Index] = "thinking" + env, err := proto.NewEnvelope(proto.TypeThinking, t.runID, proto.ThinkingPayload{ + Text: msg.Event.Delta.Thinking, + Sequence: t.seq.Add(1), + }) + if err != nil { + return translation{}, err + } + return translation{Envelopes: []proto.Envelope{env}}, nil + default: + return translation{}, nil + } +} + func (t *translator) translateSystem(line []byte) (translation, error) { var msg struct { SessionID string `json:"session_id"` @@ -150,7 +214,7 @@ func (t *translator) translateAssistant(line []byte) (translation, error) { } var envs []proto.Envelope - for _, raw := range msg.Message.Content { + for index, raw := range msg.Message.Content { var head struct { Type string `json:"type"` } @@ -159,6 +223,9 @@ func (t *translator) translateAssistant(line []byte) (translation, error) { } switch head.Type { case "text": + if t.partialBlocks[index] == "text" { + continue + } var item struct { Text string `json:"text"` } @@ -174,6 +241,9 @@ func (t *translator) translateAssistant(line []byte) (translation, error) { } envs = append(envs, env) case "thinking": + if t.partialBlocks[index] == "thinking" { + continue + } var item struct { Thinking string `json:"thinking"` } @@ -220,6 +290,10 @@ func (t *translator) translateAssistant(line []byte) (translation, error) { envs = append(envs, env) } } + // Partial flags apply only to the complete assistant frame that follows + // those stream_event deltas. Reset them so a later assistant turn that is + // delivered without partial frames is not accidentally suppressed. + clear(t.partialBlocks) return translation{Envelopes: envs}, nil } @@ -370,6 +444,8 @@ type resultUsage struct { } func (t *translator) translateResult(line []byte, subtype string) (translation, error) { + defer clear(t.partialBlocks) + var msg struct { IsError bool `json:"is_error"` Result string `json:"result"` @@ -424,7 +500,14 @@ func (t *translator) translateResult(line []byte, subtype string) (translation, // else (error_during_execution, error_max_turns, ...) is a failure. isError := msg.IsError || (subtype != "" && subtype != "success" && strings.HasPrefix(subtype, "error")) if isError { - errMsg := msg.Error + errMsg := strings.TrimSpace(msg.Error) + // Claude Code sometimes reports provider/API failures with + // subtype="success" and is_error=true, placing the useful error in + // result instead of error. Preserve that message rather than emitting + // the misleading fallback "claude_code: success". + if errMsg == "" && msg.IsError { + errMsg = strings.TrimSpace(msg.Result) + } if errMsg == "" { if subtype != "" { errMsg = "claude_code: " + subtype diff --git a/apps/parsar-daemon/internal/agent/claudecode/parser_test.go b/apps/parsar-daemon/internal/agent/claudecode/parser_test.go index 1a6fe02..839f775 100644 --- a/apps/parsar-daemon/internal/agent/claudecode/parser_test.go +++ b/apps/parsar-daemon/internal/agent/claudecode/parser_test.go @@ -110,6 +110,100 @@ func TestTranslateAssistantThinkingEmitsThinking(t *testing.T) { } } +func TestTranslatePartialMessagesStreamIncrementallyWithoutAssistantDuplicates(t *testing.T) { + tr := claudecode.NewTranslatorForTest("run_partial", nil, counterMinter()) + frames := [][]byte{ + []byte(`{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"hello "}}}`), + []byte(`{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"world"}}}`), + []byte(`{"type":"stream_event","event":{"type":"content_block_delta","index":1,"delta":{"type":"thinking_delta","thinking":"checking"}}}`), + } + + var got []proto.Envelope + for _, frame := range frames { + out, err := tr.Translate(frame) + if err != nil { + t.Fatalf("Translate partial frame: %v", err) + } + got = append(got, out.Envelopes...) + } + if len(got) != 3 { + t.Fatalf("want 3 partial envelopes, got %d", len(got)) + } + if got[0].Type != proto.TypeDelta || got[1].Type != proto.TypeDelta || got[2].Type != proto.TypeThinking { + t.Fatalf("partial envelope types = %q, %q, %q", got[0].Type, got[1].Type, got[2].Type) + } + + full, err := tr.Translate([]byte(`{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"hello world"},{"type":"thinking","thinking":"checking"}]}}`)) + if err != nil { + t.Fatalf("Translate complete assistant frame: %v", err) + } + if len(full.Envelopes) != 0 { + t.Fatalf("complete assistant frame duplicated partial content: %#v", full.Envelopes) + } + + next, err := tr.Translate([]byte(`{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"next turn without partial frames"}]}}`)) + if err != nil { + t.Fatalf("Translate next complete assistant frame: %v", err) + } + if len(next.Envelopes) != 1 || next.Envelopes[0].Type != proto.TypeDelta { + t.Fatalf("next assistant frame was suppressed by stale partial state: %#v", next.Envelopes) + } +} + +func TestTranslateStreamEventIgnoresNonTextDeltas(t *testing.T) { + tr := claudecode.NewTranslatorForTest("run_partial", nil, counterMinter()) + out, err := tr.Translate([]byte(`{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"path\":"}}}`)) + if err != nil { + t.Fatalf("Translate: %v", err) + } + if len(out.Envelopes) != 0 { + t.Fatalf("input JSON delta should not emit an envelope: %#v", out.Envelopes) + } +} + +func TestTranslatePartialMessagesSuppressOnlyMatchingContentBlock(t *testing.T) { + tr := claudecode.NewTranslatorForTest("run_partial_blocks", nil, counterMinter()) + _, err := tr.Translate([]byte(`{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"streamed"}}}`)) + if err != nil { + t.Fatalf("Translate partial frame: %v", err) + } + + out, err := tr.Translate([]byte(`{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"streamed"},{"type":"text","text":"complete-only"}]}}`)) + if err != nil { + t.Fatalf("Translate complete frame: %v", err) + } + if len(out.Envelopes) != 1 || out.Envelopes[0].Type != proto.TypeDelta { + t.Fatalf("complete-only block should be preserved: %#v", out.Envelopes) + } + got := mustDecode[struct { + Delta string `json:"delta"` + }](t, out.Envelopes[0].Payload) + if got.Delta != "complete-only" { + t.Fatalf("delta = %q, want complete-only", got.Delta) + } +} + +func TestTranslateResultClearsPartialBlocksBeforeNextTurn(t *testing.T) { + tr := claudecode.NewTranslatorForTest("run_partial_error", nil, counterMinter()) + _, err := tr.Translate([]byte(`{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"partial before failure"}}}`)) + if err != nil { + t.Fatalf("Translate partial frame: %v", err) + } + + _, err = tr.Translate([]byte(`{"type":"result","subtype":"error_during_execution","is_error":true,"error":"provider failed"}`)) + if err != nil { + t.Fatalf("Translate error result: %v", err) + } + + next, err := tr.Translate([]byte(`{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"next turn"}]}}`)) + if err != nil { + t.Fatalf("Translate next assistant frame: %v", err) + } + if len(next.Envelopes) != 1 || next.Envelopes[0].Type != proto.TypeDelta { + t.Fatalf("next assistant frame was suppressed by result state: %#v", next.Envelopes) + } +} + func TestTranslateAssistantToolUseEmitsBeforeStage(t *testing.T) { tr := claudecode.NewTranslatorForTest("run_t", nil, counterMinter()) line := []byte(`{"type":"assistant","message":{"role":"assistant","content":[ @@ -389,6 +483,21 @@ func TestTranslateResultErrorWithoutMessageFallsBackToSubtype(t *testing.T) { } } +func TestTranslateResultIsErrorSuccessSubtypeUsesResultMessage(t *testing.T) { + tr := claudecode.NewTranslatorForTest("run_99", nil, counterMinter()) + line := []byte(`{"type":"result","subtype":"success","is_error":true,"result":"API Error: 400 content rejected"}`) + out, err := tr.Translate(line) + if err != nil { + t.Fatalf("Translate: %v", err) + } + got := mustDecode[struct { + Error string `json:"error"` + }](t, out.Envelopes[0].Payload) + if got.Error != "API Error: 400 content rejected" { + t.Errorf("error text = %q", got.Error) + } +} + func TestTranslateUnknownTypeIsNoOp(t *testing.T) { tr := claudecode.NewTranslatorForTest("run_99", nil, counterMinter()) out, err := tr.Translate([]byte(`{"type":"some_future_thing","foo":1}`)) diff --git a/apps/web/src/components/admin/PairDaemonDialog.tsx b/apps/web/src/components/admin/PairDaemonDialog.tsx index 1fcc97f..1278a59 100644 --- a/apps/web/src/components/admin/PairDaemonDialog.tsx +++ b/apps/web/src/components/admin/PairDaemonDialog.tsx @@ -1,6 +1,6 @@ import { useEffect, useState } from "react" import { useTranslation } from "react-i18next" -import { Copy } from "lucide-react" +import { Check, Copy, X } from "lucide-react" import { Button } from "../ui/button" import { @@ -13,6 +13,7 @@ import { } from "../ui/dialog" import { useCreateRuntimePairing, useWorkspaceRuntimes } from "../../lib/api-runtimes" import { useBootstrapStatus } from "../../lib/api-bootstrap" +import { copyText } from "../../lib/clipboard" interface PairDaemonDialogProps { open: boolean @@ -26,12 +27,7 @@ interface PairDaemonDialogProps { onPaired?: (runtimeID: string) => void } -export function PairDaemonDialog({ - open, - onClose, - workspaceID, - onPaired, -}: PairDaemonDialogProps) { +export function PairDaemonDialog({ open, onClose, workspaceID, onPaired }: PairDaemonDialogProps) { const { t } = useTranslation("admin") const create = useCreateRuntimePairing(workspaceID) // Prefer the server's configured public URL (PARSAR_PUBLIC_URL) over the @@ -43,9 +39,11 @@ export function PairDaemonDialog({ // flips online, even when opened from a form with its own non-polling list. const listQ = useWorkspaceRuntimes(workspaceID, "agent_daemon") const [name, setName] = useState("") - const [result, setResult] = useState< - { token: string; runtimeName: string; runtimeID: string } | null - >(null) + const [result, setResult] = useState<{ + token: string + runtimeName: string + runtimeID: string + } | null>(null) const [paired, setPaired] = useState(false) const allRuntimes = listQ.data ?? [] @@ -57,8 +55,11 @@ export function PairDaemonDialog({ // re-firing on every 5s list refetch. useEffect(() => { if (!connected || !result || paired) return - setPaired(true) - onPaired?.(result.runtimeID) + const timer = window.setTimeout(() => { + setPaired(true) + onPaired?.(result.runtimeID) + }, 0) + return () => window.clearTimeout(timer) }, [connected, paired, result, onPaired]) function reset() { @@ -76,15 +77,20 @@ export function PairDaemonDialog({ const trimmed = name.trim() if (!trimmed) return const res = await create.mutateAsync({ name: trimmed, type: "agent_daemon" }) - setResult({ token: res.pairing_token, runtimeName: res.runtime.name, runtimeID: res.runtime.id }) - } - - function copyToClipboard(s: string) { - void navigator.clipboard.writeText(s) + setResult({ + token: res.pairing_token, + runtimeName: res.runtime.name, + runtimeID: res.runtime.id, + }) } return ( - { if (!o) close() }}> + { + if (!o) close() + }} + > @@ -114,12 +120,15 @@ export function PairDaemonDialog({

- {t("runtime.agentDaemon.pair.safetyTitle", { defaultValue: "How the connection works" })} + {t("runtime.agentDaemon.pair.safetyTitle", { + defaultValue: "How the connection works", + })}

  • {t("runtime.agentDaemon.pair.safetyOutbound", { - defaultValue: "This host opens an outbound connection — no inbound ports required.", + defaultValue: + "This host opens an outbound connection — no inbound ports required.", })}
  • @@ -129,15 +138,14 @@ export function PairDaemonDialog({
  • {t("runtime.agentDaemon.pair.safetyOnce", { - defaultValue: "The token is shown once — it cannot be recovered after this dialog closes.", + defaultValue: + "The token is shown once — it cannot be recovered after this dialog closes.", })}
{create.error && ( -

- {(create.error as Error).message} -

+

{(create.error as Error).message}

)} diff --git a/apps/web/src/components/conversation/StepDisplay.tsx b/apps/web/src/components/conversation/StepDisplay.tsx index 3548eae..9f61286 100644 --- a/apps/web/src/components/conversation/StepDisplay.tsx +++ b/apps/web/src/components/conversation/StepDisplay.tsx @@ -26,9 +26,19 @@ const TOOL_ICONS: Record = { glob: Search, } -function toolIcon(name: string) { +function ToolIcon({ name }: { name: string }) { const key = name.toLowerCase() - return TOOL_ICONS[key] ?? Wrench + const iconClassName = "h-3 w-3 shrink-0 text-fg-subtle" + switch (TOOL_ICONS[key]) { + case TerminalSquare: + return + case FileText: + return + case Search: + return + default: + return + } } const SUMMARY_MAX = 80 @@ -77,13 +87,18 @@ function formatElapsed(ms: number): string { // 1Hz ticker, only while `active`, so the live working card redraws // elapsed counters; stops cleanly to avoid leaking timers post-run. function useElapsedTicker(active: boolean): number { - const [, setTick] = useState(0) + const [now, setNow] = useState(0) useEffect(() => { if (!active) return - const id = window.setInterval(() => setTick((n) => n + 1), 1000) - return () => window.clearInterval(id) + const update = () => setNow(performance.now()) + const startID = window.setTimeout(update, 0) + const id = window.setInterval(update, 1000) + return () => { + window.clearTimeout(startID) + window.clearInterval(id) + } }, [active]) - return performance.now() + return now } export function StepItem({ @@ -99,7 +114,6 @@ export function StepItem({ /** Pass for completed steps; live-tick from caller for running ones. */ durationMs?: number }) { - const Icon = toolIcon(name) const upper = (name || "tool").toUpperCase() const summary = detail ? ellipsizeMiddle(detail) : "" return ( @@ -111,7 +125,7 @@ export function StepItem({ ) : ( )} - + {summary && ( - + {summary} )} @@ -144,18 +155,19 @@ export function StepItem({ export function WorkingSteps({ steps, + active, onCancel, cancelling, }: { steps: StreamingStep[] + active: boolean /** When set, render an X button next to the spinner. Parent owns the runID. */ onCancel?: () => void cancelling?: boolean }) { const { t } = useTranslation("admin") const [expanded, setExpanded] = useState(false) - const anyRunning = steps.some((s) => s.status === "running") - const now = useElapsedTicker(anyRunning) + const now = useElapsedTicker(active) const runningSteps = steps.filter((s) => s.status === "running") const completedSteps = steps.filter((s) => s.status === "completed") @@ -166,11 +178,10 @@ export function WorkingSteps({ // From first step's started_at until now (if running) or last ended_at. const firstStart = steps.length > 0 ? steps[0].started_at : null - const lastEnded = !anyRunning + const lastEnded = !active ? Math.max(...completedSteps.map((s) => s.ended_at ?? s.started_at), 0) : null - const overallMs = - firstStart === null ? 0 : (lastEnded ?? now) - firstStart + const overallMs = firstStart === null ? 0 : (lastEnded ?? now) - firstStart return (
@@ -179,9 +190,7 @@ export function WorkingSteps({ type="button" aria-expanded={expanded} aria-label={ - expanded - ? t("conversations.steps.collapseAria") - : t("conversations.steps.expandAria") + expanded ? t("conversations.steps.collapseAria") : t("conversations.steps.expandAria") } onClick={() => setExpanded((v) => !v)} className="flex shrink-0 items-center text-fg-faint transition-colors hover:text-fg-muted" @@ -193,7 +202,28 @@ export function WorkingSteps({ )}
- {total > 0 ? ( + {active ? ( +
+ + {t("conversations.steps.working")} + {completedCount > 0 && ( + + {t("conversations.steps.completedInline", { + count: completedCount, + defaultValue: "{{count}} completed", + })} + + )} + {runningCount > 0 && ( + + {t("conversations.steps.runningInline", { + count: runningCount, + defaultValue: "{{count}} running", + })} + + )} +
+ ) : total > 0 ? (
{t("conversations.steps.totalLabel", { @@ -209,14 +239,6 @@ export function WorkingSteps({ })} )} - {runningCount > 0 && ( - - {t("conversations.steps.runningInline", { - count: runningCount, - defaultValue: "{{count}} running", - })} - - )}
) : (
@@ -235,7 +257,9 @@ export function WorkingSteps({ type="button" onClick={onCancel} disabled={cancelling} - aria-label={t("conversations.steps.cancelAria", { defaultValue: "Cancel current task" })} + aria-label={t("conversations.steps.cancelAria", { + defaultValue: "Cancel current task", + })} title={t("conversations.steps.cancelAria", { defaultValue: "Cancel current task" })} className="rounded p-0.5 text-fg-faint transition-colors hover:bg-surface-muted hover:text-danger disabled:opacity-40" > @@ -292,7 +316,9 @@ export function StepTrace({ steps }: { steps: ToolStep[] }) {
)} - {/* New conversation */} + {/* Primary action */} - {/* Conversation list */} -
+ {/* Recent conversations are secondary navigation, visually separated + from the agent context and primary action above. */} +
+ + {t("conversations.sidebar.recents")} + +
+ +
{p.convsLoading ? (
{Array.from({ length: 4 }).map((_, i) => ( @@ -510,10 +516,10 @@ function ConversationSidebar(p: SidebarProps) { } }} className={cn( - "group/row relative block w-full rounded-lg border px-2.5 py-2 text-left transition-colors", + "group/row relative flex min-h-10 w-full items-center rounded-md px-2.5 py-2 text-left transition-colors", isActive - ? "border-line bg-surface shadow-sm" - : "border-transparent hover:border-line hover:bg-surface/80", + ? "bg-surface-muted text-fg" + : "text-fg-muted hover:bg-surface-muted/70 hover:text-fg", isRenaming ? "cursor-default" : "cursor-pointer", )} > @@ -576,28 +582,19 @@ function ConversationSidebar(p: SidebarProps) {
) : ( <> -
- {isActive && ( - - )} +
- {truncate(c.title || "", 18)} + {truncate(c.title || "", 24)}
-
- - {c.last_message_preview || - (c.last_message_at ? fmtAgo(c.last_message_at) : fmtAgo(c.created_at))} - -
{/* Hover-only action cluster. opacity-0 → */} {/* group-hover:opacity-100 keeps the resting state clean. */} -
+
) -} +}) function runtimeErrorViewModel( metadata: Record | undefined, diff --git a/apps/web/src/pages/admin/capabilities/AddCapabilityVersionDialog.tsx b/apps/web/src/pages/admin/capabilities/AddCapabilityVersionDialog.tsx index add8e37..4c5829d 100644 --- a/apps/web/src/pages/admin/capabilities/AddCapabilityVersionDialog.tsx +++ b/apps/web/src/pages/admin/capabilities/AddCapabilityVersionDialog.tsx @@ -29,6 +29,7 @@ import { import { Button } from "../../../components/ui/button" import { Input } from "../../../components/ui/input" import { ApiError } from "../../../lib/api-client" +import { preventDialogDismissForCredentialMenu } from "../../../lib/dialog-interactions" import { useUpdateCapability } from "../../../lib/api-capabilities" import type { Capability, CapabilityVersion } from "../../../lib/api-types" @@ -103,37 +104,40 @@ export function AddCapabilityVersionDialog({ // clobber the user's edits. useEffect(() => { if (!open) return - setName(capability.name) - setDescription(capability.description ?? "") - setSpec(null) - setInlineSecrets([]) - setRawText(prefill.rawText) - setSourceFormat(prefill.format) - setPluginUpload({ ossKey: null, uploadSource: null, validation: null }) - setSkillOssKey(null) - commitMut.reset() - updateMut.reset() + const resetTimer = window.setTimeout(() => { + setName(capability.name) + setDescription(capability.description ?? "") + setSpec(null) + setInlineSecrets([]) + setRawText(prefill.rawText) + setSourceFormat(prefill.format) + setPluginUpload({ ossKey: null, uploadSource: null, validation: null }) + setSkillOssKey(null) + commitMut.reset() + updateMut.reset() + }, 0) + return () => window.clearTimeout(resetTimer) // intentionally only on the open transition // eslint-disable-next-line react-hooks/exhaustive-deps }, [open]) - const errMsg = commitMut.error instanceof ApiError - ? commitMut.error.envelope.message - : commitMut.error instanceof Error - ? commitMut.error.message - : updateMut.error instanceof ApiError - ? updateMut.error.envelope.message - : updateMut.error instanceof Error - ? updateMut.error.message - : null + const errMsg = + commitMut.error instanceof ApiError + ? commitMut.error.envelope.message + : commitMut.error instanceof Error + ? commitMut.error.message + : updateMut.error instanceof ApiError + ? updateMut.error.envelope.message + : updateMut.error instanceof Error + ? updateMut.error.message + : null const trimmedName = name.trim() - const nameError = - !trimmedName - ? t("capabilities.errors.nameRequired") - : trimmedName.length > 50 - ? t("capabilities.errors.nameTooLong") - : null + const nameError = !trimmedName + ? t("capabilities.errors.nameRequired") + : trimmedName.length > 50 + ? t("capabilities.errors.nameTooLong") + : null // For plugin / skill-zip kinds we accept "no new upload" and let the server // reuse the previous OSS blob. So the canSubmit guard relaxes when an // inherited blob exists. @@ -141,15 +145,18 @@ export function AddCapabilityVersionDialog({ kind !== "plugin" ? true : pluginUpload.ossKey - ? pluginUpload.validation?.valid ?? false + ? (pluginUpload.validation?.valid ?? false) : !!inheritedOssLabel const skillSpecReady = kind !== "skill" ? true - : !!skillOssKey || !!inheritedOssLabel || (!!spec && isImportSpecReady(kind, spec, inlineSecrets)) + : !!skillOssKey || + !!inheritedOssLabel || + (!!spec && isImportSpecReady(kind, spec, inlineSecrets)) - const mcpSpecReady = kind !== "mcp" ? true : !!spec && isImportSpecReady(kind, spec, inlineSecrets) + const mcpSpecReady = + kind !== "mcp" ? true : !!spec && isImportSpecReady(kind, spec, inlineSecrets) const canSubmit = !commitMut.isPending && @@ -194,13 +201,13 @@ export function AddCapabilityVersionDialog({ const ossKeyToSend = kind === "plugin" - ? pluginUpload.ossKey ?? undefined + ? (pluginUpload.ossKey ?? undefined) : kind === "skill" - ? skillOssKey ?? undefined + ? (skillOssKey ?? undefined) : undefined const uploadSourceToSend = kind === "plugin" - ? pluginUpload.uploadSource ?? undefined + ? (pluginUpload.uploadSource ?? undefined) : kind === "skill" && skillOssKey ? "zip" : undefined @@ -208,9 +215,7 @@ export function AddCapabilityVersionDialog({ const payload: ImportCapabilityVersionCommitRequest = { canonical_spec: fallbackSpec, inline_secrets: kind === "plugin" || inlineSecrets.length === 0 ? undefined : inlineSecrets, - source_payload: rawText - ? { raw_text: rawText, source_format: sourceFormat } - : undefined, + source_payload: rawText ? { raw_text: rawText, source_format: sourceFormat } : undefined, // omit oss_key on plugin/skill-zip reuse — backend treats missing key // as "carry forward the previous version's blob". oss_key: ossKeyToSend, @@ -235,24 +240,21 @@ export function AddCapabilityVersionDialog({ { - if (commitMut.isPending || updateMut.isPending) e.preventDefault() - }} + onInteractOutside={preventDialogDismissForCredentialMenu} > {t("capabilities.versions.add.title", { name: capability.name })} - - {t("capabilities.versions.add.description")} - + {t("capabilities.versions.add.description")} {prefill.didPrefill && ( {t("capabilities.versions.add.prefillFromLatest", { version: latestVersion?.version ?? "", - defaultValue: "Pre-filled with the previous version ({{version}}). Edits will be submitted as a new version.", + defaultValue: + "Pre-filled with the previous version ({{version}}). Edits will be submitted as a new version.", })} )} @@ -260,17 +262,17 @@ export function AddCapabilityVersionDialog({ {t("capabilities.versions.add.reuseExistingZip", { filename: inheritedOssLabel, - defaultValue: "Current version package: {{filename}}. If you do not re-upload, the new version will reuse this package.", + defaultValue: + "Current version package: {{filename}}. If you do not re-upload, the new version will reuse this package.", })} )} {inheritedInlineSecrets.length > 0 && ( {t("capabilities.versions.add.inlineSecretLostWarning", { - keys: inheritedInlineSecrets - .map((e) => `${e.server}.${e.envKey}`) - .join(", "), - defaultValue: "Previous-version inline secrets ({{keys}}) are hidden. Re-enter them in plaintext to keep, or switch to managed credentials.", + keys: inheritedInlineSecrets.map((e) => `${e.server}.${e.envKey}`).join(", "), + defaultValue: + "Previous-version inline secrets ({{keys}}) are hidden. Re-enter them in plaintext to keep, or switch to managed credentials.", })} )} @@ -388,17 +390,14 @@ function usePrefillFromLatest(latestVersion: CapabilityVersion | undefined): { } { return useMemo(() => { const sp = latestVersion?.source_payload as - | { raw_text?: string; source_format?: string; format?: string; body?: string } - | undefined + { raw_text?: string; source_format?: string; format?: string; body?: string } | undefined // Accepts two source_payload shapes for forward-compat: // { raw_text, source_format } — new dialog // { format, body } — early server code const rawText = sp?.raw_text ?? sp?.body ?? "" const fmtRaw = (sp?.source_format ?? sp?.format ?? "").toLowerCase() const valid: SourceFormat[] = ["json", "toml", "markdown"] - const format = (valid as string[]).includes(fmtRaw) - ? (fmtRaw as SourceFormat) - : "json" + const format = (valid as string[]).includes(fmtRaw) ? (fmtRaw as SourceFormat) : "json" return { rawText, format, didPrefill: rawText.length > 0 } }, [latestVersion]) } diff --git a/apps/web/src/pages/admin/capabilities/CredentialKindCombobox.tsx b/apps/web/src/pages/admin/capabilities/CredentialKindCombobox.tsx index 3db8d77..ee224e9 100644 --- a/apps/web/src/pages/admin/capabilities/CredentialKindCombobox.tsx +++ b/apps/web/src/pages/admin/capabilities/CredentialKindCombobox.tsx @@ -58,11 +58,12 @@ export function CredentialKindCombobox({ ) }, [items, search]) - const errMsg = kindsQ.error instanceof ApiError - ? kindsQ.error.envelope.message - : kindsQ.error instanceof Error - ? kindsQ.error.message - : null + const errMsg = + kindsQ.error instanceof ApiError + ? kindsQ.error.envelope.message + : kindsQ.error instanceof Error + ? kindsQ.error.message + : null return ( <> @@ -71,11 +72,7 @@ export function CredentialKindCombobox({ diff --git a/scripts/dev-server-up.sh b/scripts/dev-server-up.sh index bfcb911..870863c 100755 --- a/scripts/dev-server-up.sh +++ b/scripts/dev-server-up.sh @@ -203,6 +203,7 @@ TMUX_ENV="PARSAR_ADDR=:${PORT} DATABASE_URL='${DATABASE_URL}' PARSAR_DEV_AUTH=${ [[ -n "${AGENT_DAEMON_SANDBOX_TTL:-}" ]] && TMUX_ENV+=" AGENT_DAEMON_SANDBOX_TTL='${AGENT_DAEMON_SANDBOX_TTL}'" [[ -n "${AGENT_DAEMON_SANDBOX_AUTO_RENEW:-}" ]] && TMUX_ENV+=" AGENT_DAEMON_SANDBOX_AUTO_RENEW='${AGENT_DAEMON_SANDBOX_AUTO_RENEW}'" [[ -n "${AGENT_DAEMON_SANDBOX_TTL_HOURS:-}" ]] && TMUX_ENV+=" AGENT_DAEMON_SANDBOX_TTL_HOURS='${AGENT_DAEMON_SANDBOX_TTL_HOURS}'" +[[ -n "${PARSAR_DAEMON_BINARY_DIR:-}" ]] && TMUX_ENV+=" PARSAR_DAEMON_BINARY_DIR='${PARSAR_DAEMON_BINARY_DIR}'" tmux new-session -d -s "${TMUX_SESSION}" \ "${TMUX_ENV} '${BIN_PATH}' 2>&1 | tee '${LOG_PATH}'" diff --git a/server/internal/dev/run_stream.go b/server/internal/dev/run_stream.go index 5e02d36..a17f871 100644 --- a/server/internal/dev/run_stream.go +++ b/server/internal/dev/run_stream.go @@ -576,15 +576,17 @@ func stringFromMap(values map[string]any, key string) string { } } -func persistFinal(ctx context.Context, runtimeStore RuntimeStore, streamStore runStreamStore, runID string, source string, final *connector.PromptOutput) { +func persistFinal(_ context.Context, runtimeStore RuntimeStore, streamStore runStreamStore, runID string, source string, final *connector.PromptOutput) { + terminalCtx, terminalCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer terminalCancel() if final == nil { final = &connector.PromptOutput{Content: ""} } - _, err := streamStore.SendAssistantMessageFromRun(ctx, store.SendAssistantMessageFromRunInput{RunID: runID, Source: source, Content: final.Content, Transcript: final.Transcript, Usage: final.Usage}) + _, err := streamStore.SendAssistantMessageFromRun(terminalCtx, store.SendAssistantMessageFromRunInput{RunID: runID, Source: source, Content: final.Content, Transcript: final.Transcript, Usage: final.Usage}) if err != nil { log.Bg().Warn("persist streamed agent final failed", "run_id", runID, "error", err) // failRunWithVisibleMessage records run.failed internally. - failRunWithVisibleMessage(ctx, runtimeStore, runID, source, err.Error()) + failRunWithVisibleMessage(terminalCtx, runtimeStore, runID, source, err.Error()) } } @@ -594,8 +596,10 @@ func persistFinal(ctx context.Context, runtimeStore RuntimeStore, streamStore ru // // The in-band path (EventError / empty EventDone) goes through // eventPersistencePayload + failRunRowOnly to avoid double-emit. -func failRunWithVisibleMessage(ctx context.Context, runtimeStore RuntimeStore, runID string, source string, reason string) { - if err := runtimeStore.FailAgentRun(ctx, store.FailAgentRunInput{RunID: runID, Source: source, Reason: reason}); err != nil { +func failRunWithVisibleMessage(_ context.Context, runtimeStore RuntimeStore, runID string, source string, reason string) { + terminalCtx, terminalCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer terminalCancel() + if err := runtimeStore.FailAgentRun(terminalCtx, store.FailAgentRunInput{RunID: runID, Source: source, Reason: reason}); err != nil { log.Bg().Warn("failed to mark streamed agent run failed", "run_id", runID, "error", err) return } @@ -617,15 +621,19 @@ func failRunWithVisibleMessage(ctx context.Context, runtimeStore RuntimeStore, r // lifecycle event. Used by the in-band stream-error path where // eventPersistencePayload has already emitted run.failed from the // connector's terminal frame. -func failRunRowOnly(ctx context.Context, runtimeStore RuntimeStore, runID string, source string, reason string) { - if err := runtimeStore.FailAgentRun(ctx, store.FailAgentRunInput{RunID: runID, Source: source, Reason: reason}); err != nil { +func failRunRowOnly(_ context.Context, runtimeStore RuntimeStore, runID string, source string, reason string) { + terminalCtx, terminalCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer terminalCancel() + if err := runtimeStore.FailAgentRun(terminalCtx, store.FailAgentRunInput{RunID: runID, Source: source, Reason: reason}); err != nil { log.Bg().Warn("failed to mark streamed agent run failed", "run_id", runID, "error", err) } } -func publishAndFailRun(ctx context.Context, runtimeStore RuntimeStore, broker interface { +func publishAndFailRun(_ context.Context, runtimeStore RuntimeStore, broker interface { Publish(string, connector.PromptEvent) }, runID string, source string, err error) { + terminalCtx, terminalCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer terminalCancel() msg := err.Error() errorEvent := connector.PromptEvent{Type: connector.EventError, Error: msg} doneEvent := connector.PromptEvent{Type: connector.EventDone, Final: &connector.PromptOutput{Content: "", Metadata: map[string]any{"source": source, "error": msg}}} @@ -635,10 +643,10 @@ func publishAndFailRun(ctx context.Context, runtimeStore RuntimeStore, broker in // EventDone-empty already records run.failed via // eventPersistencePayload; use failRunRowOnly to avoid a // duplicate event row. - _ = recordPromptEvent(ctx, streamStore, runID, errorEvent) - _ = recordPromptEvent(ctx, streamStore, runID, doneEvent) + _ = recordPromptEvent(terminalCtx, streamStore, runID, errorEvent) + _ = recordPromptEvent(terminalCtx, streamStore, runID, doneEvent) } - failRunRowOnly(ctx, runtimeStore, runID, source, msg) + failRunRowOnly(terminalCtx, runtimeStore, runID, source, msg) } func conversationRunParams(w http.ResponseWriter, r *http.Request) (string, string, bool) { diff --git a/server/internal/dev/run_stream_test.go b/server/internal/dev/run_stream_test.go index 49e20c5..0d6671f 100644 --- a/server/internal/dev/run_stream_test.go +++ b/server/internal/dev/run_stream_test.go @@ -17,6 +17,28 @@ import ( "github.com/jackc/pgx/v5/pgxpool" ) +type terminalContextRuntimeStore struct { + stubRuntimeStore + ctxErr error +} + +func (s *terminalContextRuntimeStore) FailAgentRun(ctx context.Context, _ store.FailAgentRunInput) error { + s.ctxErr = ctx.Err() + return nil +} + +func TestFailRunRowOnlyUsesFreshContextAfterDispatchTimeout(t *testing.T) { + dispatchCtx, cancel := context.WithCancel(context.Background()) + cancel() + runtimeStore := &terminalContextRuntimeStore{} + + failRunRowOnly(dispatchCtx, runtimeStore, testRunID, "conversation_stream", context.DeadlineExceeded.Error()) + + if runtimeStore.ctxErr != nil { + t.Fatalf("FailAgentRun context error = %v, want nil", runtimeStore.ctxErr) + } +} + func TestConversationRunStreamStartAndLateReplayPersistsFinal(t *testing.T) { db := openDevRouteTestDB(t) ctx := context.Background() diff --git a/server/internal/store/store.go b/server/internal/store/store.go index 8cfe74e..8e8b3b6 100644 --- a/server/internal/store/store.go +++ b/server/internal/store/store.go @@ -7005,16 +7005,10 @@ func (s *Store) SendUserMessageToConversation(ctx context.Context, input SendUse return result, err } - mentionNames := mentionPattern.FindAllString(content, -1) + mentionNames := userMessageMentionNames(content) if len(input.MentionedAgentIDs) > 0 { mentionNames = nil } - // 1v1 fallback: when no @-mention and no explicit MentionedAgentIDs, route to the - // conversation's bound primary_agent so a typed message reaches it (ChatGPT-style). - implicitPrimary := "" - if len(mentionNames) == 0 && len(input.MentionedAgentIDs) == 0 { - implicitPrimary = strings.TrimSpace(conversation.PrimaryAgentID) - } mentionedAgents := make([]mentionedAgent, 0, len(input.MentionedAgentIDs)+len(mentionNames)+1) seenAgents := map[string]struct{}{} for _, mention := range mentionNames { @@ -7051,8 +7045,14 @@ func (s *Store) SendUserMessageToConversation(ctx context.Context, input SendUse seenAgents[agent.agentID] = struct{}{} mentionedAgents = append(mentionedAgents, agent) } - // Implicit primary_agent fallback: must be active, otherwise silently drop to "no run - // dispatched" so the user message still lands and the UI shows the bound-agent-disabled state. + // Only an actual agent mention suppresses the 1v1 fallback. The mention parser + // excludes @ fragments embedded in email addresses and SSH repository URLs. + implicitPrimary := "" + if len(mentionNames) == 0 && len(input.MentionedAgentIDs) == 0 { + implicitPrimary = strings.TrimSpace(conversation.PrimaryAgentID) + } + // The implicit primary must be active; otherwise silently drop to "no run dispatched" + // so the user message still lands and the UI shows the bound-agent-disabled state. if implicitPrimary != "" { agentUUID, err := uuid(implicitPrimary) if err == nil { @@ -7656,6 +7656,19 @@ type mentionedAgent struct { var mentionPattern = regexp.MustCompile(`@[\p{Han}A-Za-z0-9_-]+`) +var userMessageMentionPattern = regexp.MustCompile(`(?:^|[\s\(\[\{，。！？、])@([\p{Han}A-Za-z0-9_-]+)`) + +func userMessageMentionNames(content string) []string { + matches := userMessageMentionPattern.FindAllStringSubmatch(content, -1) + names := make([]string, 0, len(matches)) + for _, match := range matches { + if len(match) == 2 { + names = append(names, "@"+match[1]) + } + } + return names +} + var ansiEscapePattern = regexp.MustCompile(`\x1b\[[0-9;?]*[a-zA-Z]`) var buildLinePattern = regexp.MustCompile(`(?m)^>\s*build.*$`) var shellPromptLinePattern = regexp.MustCompile(`(?m)^\$\s.*$`) diff --git a/server/internal/store/store_test.go b/server/internal/store/store_test.go index 7e4f3b8..333b897 100644 --- a/server/internal/store/store_test.go +++ b/server/internal/store/store_test.go @@ -3646,6 +3646,44 @@ func TestSendUserMessageImplicitPrimaryAgentDispatchesWithoutMention(t *testing. t.Fatalf("explicit @primary + bound primary should still be 1 run, got %d", len(sent2.RunIDs)) } + // An SSH repository URL contains an @ but is not an agent mention. It must + // still route to the bound primary agent. + sentSSH, err := store.SendUserMessageToConversation(ctx, SendUserMessageToConversationInput{ + ConversationID: conv.ID, + UserID: ids.UserID, + Content: "git@github.com:sandbaseai/deepseek-harness-handbook.git", + }) + if err != nil { + t.Fatal(err) + } + if len(sentSSH.RunIDs) != 1 { + t.Fatalf("SSH URL should dispatch to the implicit primary, got %d runs", len(sentSSH.RunIDs)) + } + + sentEmail, err := store.SendUserMessageToConversation(ctx, SendUserMessageToConversationInput{ + ConversationID: conv.ID, + UserID: ids.UserID, + Content: "contact david@example.com for details", + }) + if err != nil { + t.Fatal(err) + } + if len(sentEmail.RunIDs) != 1 { + t.Fatalf("email address should dispatch to the implicit primary, got %d runs", len(sentEmail.RunIDs)) + } + + sentUnknownMention, err := store.SendUserMessageToConversation(ctx, SendUserMessageToConversationInput{ + ConversationID: conv.ID, + UserID: ids.UserID, + Content: "@missing-agent please handle this", + }) + if err != nil { + t.Fatal(err) + } + if len(sentUnknownMention.RunIDs) != 0 { + t.Fatalf("unknown explicit mention should not fall back to primary, got %d runs", len(sentUnknownMention.RunIDs)) + } + // Unbound conversation + bare message → no dispatch. convNoPrimary, err := store.CreateWorkspaceConversation(ctx, CreateWorkspaceConversationInput{ WorkspaceID: ids.WorkspaceID,