From db406e332d4d268fd194bf9d926fa1c386457043 Mon Sep 17 00:00:00 2001 From: Hazem Nureldin Date: Tue, 4 Aug 2026 08:20:13 +0100 Subject: [PATCH 01/24] fix:lohs retrival using wss --- frontend/.devcontainer/devcontainer-lock.json | 14 ++ .../src/routes/SingleWorkflowPage.tsx | 17 +- .../lib/components/BaseWorkflowRelay.tsx | 45 +++- .../lib/components/RelayEnvironment.ts | 51 +++-- .../lib/components/TasksFlow.tsx | 22 +- frontend/relay-workflows-lib/lib/main.ts | 1 + .../lib/views/BaseSingleWorkflowView.tsx | 164 +++++++++++--- .../lib/views/SingleWorkflowView.tsx | 5 + .../lib/views/TaskLogViewer.tsx | 212 ++++++++++++++++++ 9 files changed, 475 insertions(+), 56 deletions(-) create mode 100644 frontend/.devcontainer/devcontainer-lock.json create mode 100644 frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx diff --git a/frontend/.devcontainer/devcontainer-lock.json b/frontend/.devcontainer/devcontainer-lock.json new file mode 100644 index 000000000..f1bff2f3e --- /dev/null +++ b/frontend/.devcontainer/devcontainer-lock.json @@ -0,0 +1,14 @@ +{ + "features": { + "ghcr.io/devcontainers/features/common-utils:2.4.2": { + "version": "2.4.2", + "resolved": "ghcr.io/devcontainers/features/common-utils@sha256:bebfdcd6097a35506bf0f064a31e52ad4205467d9d7a226a688f51b851c88b65", + "integrity": "sha256:bebfdcd6097a35506bf0f064a31e52ad4205467d9d7a226a688f51b851c88b65" + }, + "ghcr.io/devcontainers/features/git-lfs:1.2.3": { + "version": "1.2.3", + "resolved": "ghcr.io/devcontainers/features/git-lfs@sha256:7acbf0a99325949b6a2906ebf5aa421dad72b89ab8045031a60e69cb394a16b1", + "integrity": "sha256:7acbf0a99325949b6a2906ebf5aa421dad72b89ab8045031a60e69cb394a16b1" + } + } +} diff --git a/frontend/dashboard/src/routes/SingleWorkflowPage.tsx b/frontend/dashboard/src/routes/SingleWorkflowPage.tsx index 9a0fa0f36..c85a32fe8 100644 --- a/frontend/dashboard/src/routes/SingleWorkflowPage.tsx +++ b/frontend/dashboard/src/routes/SingleWorkflowPage.tsx @@ -1,13 +1,14 @@ import { Container, Box, Typography } from "@mui/material"; import { useParams, Link, useSearchParams } from "react-router-dom"; -import { Suspense, useMemo } from "react"; +import { Suspense, useMemo, useState } from "react"; import "react-resizable/css/styles.css"; import { Breadcrumbs } from "@diamondlightsource/sci-react-ui"; -import { SingleWorkflowView, WorkflowsNavbar } from "relay-workflows-lib"; + import { visitTextToVisit, WorkflowErrorBoundaryWithRetry, } from "workflows-lib"; +import { SingleWorkflowView, WorkflowsNavbar, TaskLogViewer } from "relay-workflows-lib"; function SingleWorkflowPage() { const { visitid, workflowName } = useParams<{ @@ -17,6 +18,7 @@ function SingleWorkflowPage() { const [searchParams] = useSearchParams(); const taskParam = searchParams.get("tasks"); + const [selectedTaskId, setSelectedTaskId] = useState(null); if (visitid) { localStorage.setItem("instrumentSessionID", visitid); @@ -64,6 +66,14 @@ function SingleWorkflowPage() { visit={visit} workflowName={workflowName} taskIds={taskIds} + onSelectTask={(taskId: string) => setSelectedTaskId(taskId)} + /> + + {/* Real-time Task Log Viewer */} + )} @@ -79,11 +89,10 @@ function SingleWorkflowPage() { mb={4} > No valid workflow selected - {/* Go to instrumentSession or home page */} )} ); } -export default SingleWorkflowPage; +export default SingleWorkflowPage; \ No newline at end of file diff --git a/frontend/relay-workflows-lib/lib/components/BaseWorkflowRelay.tsx b/frontend/relay-workflows-lib/lib/components/BaseWorkflowRelay.tsx index fb92e1a42..42d9f953e 100644 --- a/frontend/relay-workflows-lib/lib/components/BaseWorkflowRelay.tsx +++ b/frontend/relay-workflows-lib/lib/components/BaseWorkflowRelay.tsx @@ -9,7 +9,7 @@ import { useParams, useNavigate } from "react-router-dom"; import { graphql } from "relay-runtime"; import { useFragment } from "react-relay"; import { BaseWorkflowRelayFragment$key } from "./__generated__/BaseWorkflowRelayFragment.graphql"; -import TasksFlow from "./TasksFlow"; +import TasksFlow from "./TasksFlow"; export const BaseWorkflowRelayFragment = graphql` fragment BaseWorkflowRelayFragment on Workflow { @@ -35,6 +35,7 @@ interface BaseWorkflowRelayProps { expanded?: boolean; onChange?: () => void; fragmentRef: BaseWorkflowRelayFragment$key; + onSelectTask?: (taskId: string) => void; } export default function BaseWorkflowRelay({ @@ -43,20 +44,35 @@ export default function BaseWorkflowRelay({ expanded, onChange, fragmentRef, + onSelectTask, }: BaseWorkflowRelayProps) { const { workflowName: workflowNameURL } = useParams<{ workflowName: string; }>(); + const navigate = useNavigate(); - const data = useFragment(BaseWorkflowRelayFragment, fragmentRef); + + const data = useFragment( + BaseWorkflowRelayFragment, + fragmentRef, + ); + const statusText = data.status?.__typename ?? "Unknown"; - const [selectedTaskIds, setSelectedTaskIds] = useSelectedTaskIds(); + + const [selectedTaskIds, setSelectedTaskIds] = + useSelectedTaskIds(); const onNavigate = React.useCallback( (taskId: string, event?: React.MouseEvent) => { - const isCtrl = event?.ctrlKey || event?.metaKey; + const isCtrl = + event?.ctrlKey || + event?.metaKey; let updatedTaskIds: string[]; + console.log( + "TASK CLICKED", + taskId + ); if (isCtrl) { updatedTaskIds = selectedTaskIds.includes(taskId) @@ -65,12 +81,25 @@ export default function BaseWorkflowRelay({ } else { updatedTaskIds = [taskId]; } + if (workflowNameURL !== data.name) { - void navigate(`/workflows/${visitToText(data.visit)}/${data.name}`); + void navigate( + `/workflows/${visitToText(data.visit)}/${data.name}`, + ); + } + + if (onSelectTask) { + onSelectTask(taskId); } - setSelectedTaskIds(updatedTaskIds); }, - [navigate, selectedTaskIds, setSelectedTaskIds, workflowNameURL, data], + [ + navigate, + selectedTaskIds, + setSelectedTaskIds, + workflowNameURL, + data, + onSelectTask, + ], ); return ( @@ -127,4 +156,4 @@ export default function BaseWorkflowRelay({ ); -} +} \ No newline at end of file diff --git a/frontend/relay-workflows-lib/lib/components/RelayEnvironment.ts b/frontend/relay-workflows-lib/lib/components/RelayEnvironment.ts index 40bbeaf47..571bca628 100644 --- a/frontend/relay-workflows-lib/lib/components/RelayEnvironment.ts +++ b/frontend/relay-workflows-lib/lib/components/RelayEnvironment.ts @@ -77,7 +77,7 @@ const fetchFn: FetchFunction = async (request, variables) => { const resp = await fetch(HTTP_ENDPOINT, { method: "POST", headers, - credentials: "include", + // credentials: "include", body: JSON.stringify({ query: request.text, // <-- The GraphQL document composed by Relay variables, @@ -91,9 +91,25 @@ const fetchFn: FetchFunction = async (request, variables) => { return await resp.json(); // eslint-disable-line @typescript-eslint/no-unsafe-return }; - +console.log("HTTP_ENDPOINTXXXXXXXXXXXXXXXXXXXXXXXXXXXX:", HTTP_ENDPOINT); +console.log("WS_ENDPOINTYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYY:", WS_ENDPOINT); export const wsClient = createClient({ url: WS_ENDPOINT, + on: { + connecting: () => console.log("WS connecting"), + opened: () => console.log("WS opened"), + connected: () => console.log("WS connected"), + closed: (event) => console.log("WS closed", event), + }, + webSocketImpl: class extends WebSocket { + constructor(url: string | URL, protocols?: string | string[]) { + console.log("Creating browser WebSocket:", url); + super(url, protocols); + } + }, + + + connectionParams: async () => { if (!USE_AUTH_GATEWAY && !keycloak.authenticated) { await ensureKeycloakInit(); @@ -108,6 +124,12 @@ export const wsClient = createClient({ }); const subscribeFn: SubscribeFunction = (operation, variables) => { + console.log( + "WS SUBSCRIBE STARTED:", + operation.name, + variables + ); + return Observable.create((sink) => { const cleanup = wsClient.subscribe( { @@ -117,20 +139,23 @@ const subscribeFn: SubscribeFunction = (operation, variables) => { }, { next: (response) => { - const data = response.data; - if (data) { - sink.next({ data } as GraphQLResponse); - } else if (data == null) { - console.warn("Data is null:", response); - } else { - console.error("Subscription error response:", response); - sink.error(new Error("Subscription response missing data")); - } + console.log("WS SUBSCRIPTION RESPONSE:", response); + + sink.next(response as GraphQLResponse); + }, + + error: (error) => { + console.error("WS SUBSCRIPTION ERROR:", error); + sink.error(error); + }, + + complete: () => { + console.log("WS SUBSCRIPTION COMPLETE"); + sink.complete(); }, - error: sink.error.bind(sink), - complete: sink.complete.bind(sink), }, ); + return cleanup; }); }; diff --git a/frontend/relay-workflows-lib/lib/components/TasksFlow.tsx b/frontend/relay-workflows-lib/lib/components/TasksFlow.tsx index d83256560..dc2506b8d 100644 --- a/frontend/relay-workflows-lib/lib/components/TasksFlow.tsx +++ b/frontend/relay-workflows-lib/lib/components/TasksFlow.tsx @@ -144,30 +144,42 @@ const TasksFlow: React.FC = ({ useEffect(() => { const handleResizeAndOverflow = () => { - if (containerRef.current) { - const { width, height } = containerRef.current.getBoundingClientRect(); - const boundingBox = getNodesBounds(layoutedNodes); - setIsOverflow(boundingBox.width > width || boundingBox.height > height); + if (containerRef.current && reactFlowInstance.current) { + const { width, height } = + containerRef.current.getBoundingClientRect(); + + const nodeLookup = reactFlowInstance.current.getNodes(); + + const boundingBox = getNodesBounds(nodeLookup); + + setIsOverflow( + boundingBox.width > width || + boundingBox.height > height + ); } }; + const resizeObserver = new ResizeObserver(handleResizeAndOverflow); + const currentContainerRef = containerRef.current; if (currentContainerRef) { resizeObserver.observe(currentContainerRef); } + handleResizeAndOverflow(); + window.addEventListener("resize", handleResizeAndOverflow); return () => { if (currentContainerRef) { resizeObserver.unobserve(currentContainerRef); } + resizeObserver.disconnect(); window.removeEventListener("resize", handleResizeAndOverflow); }; }, [layoutedNodes, layoutedEdges]); - return ( (null); - const taskTree = useMemo(() => buildTaskTree(fetchedTasks), [fetchedTasks]); + const [ + selectedTaskIds, + setSelectedTaskIds, + ] = useSelectedTaskIds(); + + const [ + filledTaskId, + setFilledTaskId, + ] = useState(null); + + + // The task currently opened in the log viewer + const [ + selectedTaskId, + setSelectedTaskId, + ] = useState(null); + + + const taskTree = useMemo( + () => buildTaskTree(fetchedTasks), + [fetchedTasks] + ); + const outputTaskIds: string[] = useMemo(() => { const newOutputTaskIds: string[] = []; + const traverse = (tasks: TaskNode[]) => { - const sortedTasks = [...tasks].sort((a, b) => a.id.localeCompare(b.id)); + const sortedTasks = [...tasks].sort( + (a, b) => a.id.localeCompare(b.id) + ); + sortedTasks.forEach((taskNode) => { + if ( taskNode.children && taskNode.children.length === 0 && !newOutputTaskIds.includes(taskNode.id) ) { newOutputTaskIds.push(taskNode.id); - } else if (taskNode.children && taskNode.children.length > 0) { + } + + else if ( + taskNode.children && + taskNode.children.length > 0 + ) { traverse(taskNode.children); } + }); }; + traverse(taskTree); + return newOutputTaskIds; + }, [taskTree]); + const handleSelectOutput = () => { setSelectedTaskIds(outputTaskIds); }; + const handleSelectClear = () => { setSelectedTaskIds([]); + setSelectedTaskId(null); }; + const onArtifactHover = useCallback( (artifact: Artifact | null) => { - setFilledTaskId(artifact ? artifact.parentTaskId : null); + setFilledTaskId( + artifact + ? artifact.parentTaskId + : null + ); }, - [setFilledTaskId], + [] ); + useEffect(() => { setSelectedTaskIds(taskIds ?? []); - }, [taskIds, setSelectedTaskIds]); + }, [ + taskIds, + setSelectedTaskIds, + ]); + + const artifactList: Artifact[] = useMemo(() => { + const filteredTasks = selectedTaskIds.length + ? selectedTaskIds - .map((id) => fetchedTasks.find((task) => task.id === id)) - .filter((task): task is Task => !!task) + .map((id) => + fetchedTasks.find( + (task) => task.id === id + ) + ) + .filter( + (task): task is Task => + !!task + ) + : fetchedTasks; - return filteredTasks.flatMap((task) => task.artifacts); - }, [selectedTaskIds, fetchedTasks]); + + + return filteredTasks.flatMap( + (task) => task.artifacts + ); + + }, [ + selectedTaskIds, + fetchedTasks, + ]); + + if (!data || !data.status) { return null; } + + return ( <> + + + + OUTPUT + + CLEAR + - {fragmentRef && ( - - )} + + + + + + + + + {taskIds && ( )} - {} + + + + {/* */} + + + + + ); -} +} \ No newline at end of file diff --git a/frontend/relay-workflows-lib/lib/views/SingleWorkflowView.tsx b/frontend/relay-workflows-lib/lib/views/SingleWorkflowView.tsx index 3aed782dc..c30e7982b 100644 --- a/frontend/relay-workflows-lib/lib/views/SingleWorkflowView.tsx +++ b/frontend/relay-workflows-lib/lib/views/SingleWorkflowView.tsx @@ -23,6 +23,7 @@ export interface SingleWorkflowViewProps { workflowName: string; taskIds?: string[]; onNullSubscriptionData?: () => void; + onSelectTask?: (taskId: string) => void; } export default function SingleWorkflowView(props: SingleWorkflowViewProps) { @@ -31,6 +32,10 @@ export default function SingleWorkflowView(props: SingleWorkflowViewProps) { { visit: props.visit, name: props.workflowName, + + + + }, ); const finished = diff --git a/frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx b/frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx new file mode 100644 index 000000000..c22f7cc50 --- /dev/null +++ b/frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx @@ -0,0 +1,212 @@ +import React, { + useEffect, + useMemo, + useRef, + useState, +} from "react"; +import { + Box, + Paper, + Typography, + CircularProgress, +} from "@mui/material"; +import { graphql, useSubscription } from "react-relay"; +import { GraphQLSubscriptionConfig } from "relay-runtime"; +import { Visit } from "@diamondlightsource/sci-react-ui"; +import { TaskLogViewerSubscription } from "./__generated__/TaskLogViewerSubscription.graphql"; + + +const taskLogViewerSubscription = graphql` + subscription TaskLogViewerSubscription( + $visit: VisitInput! + $workflowName: String! + $taskId: String! + ) { + logs( + visit: $visit + workflowName: $workflowName + taskId: $taskId + ) { + content + podName + } + } +`; + + +interface TaskLogViewerProps { + visit: Visit; + workflowName: string; + selectedTaskId: string | null; +} + + +export const TaskLogViewer: React.FC = ({ + visit, + workflowName, + selectedTaskId, +}) => { + + const [logLines, setLogLines] = useState([]); + + const containerRef = useRef(null); + + + useEffect(() => { + console.log("TaskLogViewer selection changed:", { + workflowName, + selectedTaskId, + visit, + }); + + setLogLines([]); + + }, [ + selectedTaskId, + workflowName, + visit, + ]); + + + + const subscriptionConfig = + useMemo>( + () => ({ + subscription: taskLogViewerSubscription, + + variables: { + visit, + workflowName, + taskId: selectedTaskId ?? "", + }, + + onNext: (payload) => { + console.log("LOG EVENT:", payload); + + const line = payload?.logs?.content; + + if (line) { + setLogLines((prev) => [ + ...prev, + line, + ]); + } + }, + + onError: (error) => { + console.error("Log subscription error:", error); + }, + }), + [ + visit, + workflowName, + selectedTaskId, + ], + ); + + + console.log("SUBSCRIBING WITH:", subscriptionConfig.variables); + // IMPORTANT: + // This hook must ALWAYS run + useSubscription(subscriptionConfig); + + + + useEffect(() => { + + if (containerRef.current) { + containerRef.current.scrollTop = + containerRef.current.scrollHeight; + } + + }, [logLines]); + + + + return ( + + + + + + {selectedTaskId ?? "No task selected"} + + + + {selectedTaskId && ( + + )} + + + + + + + + {!selectedTaskId && ( + + Select a task. + + )} + + + {selectedTaskId && + logLines.length === 0 && ( + + Waiting for log output... + + )} + + + {logLines.map((line,index)=>( + + {line} + + ))} + + + + + ); +}; + + +export default TaskLogViewer; \ No newline at end of file From 160435b9f62458fd5e92dbe46510ddf033069577 Mon Sep 17 00:00:00 2001 From: Hazem Nureldin Date: Tue, 4 Aug 2026 10:24:28 +0100 Subject: [PATCH 02/24] fix:taskname added, log seletion done --- frontend/dashboard/src/routes/SingleWorkflowPage.tsx | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/frontend/dashboard/src/routes/SingleWorkflowPage.tsx b/frontend/dashboard/src/routes/SingleWorkflowPage.tsx index c85a32fe8..66459b8b6 100644 --- a/frontend/dashboard/src/routes/SingleWorkflowPage.tsx +++ b/frontend/dashboard/src/routes/SingleWorkflowPage.tsx @@ -18,7 +18,6 @@ function SingleWorkflowPage() { const [searchParams] = useSearchParams(); const taskParam = searchParams.get("tasks"); - const [selectedTaskId, setSelectedTaskId] = useState(null); if (visitid) { localStorage.setItem("instrumentSessionID", visitid); @@ -66,15 +65,14 @@ function SingleWorkflowPage() { visit={visit} workflowName={workflowName} taskIds={taskIds} - onSelectTask={(taskId: string) => setSelectedTaskId(taskId)} /> {/* Real-time Task Log Viewer */} - + /> */} )} From b9b7ca17b8187be5adba7026d3a6afcad81f946c Mon Sep 17 00:00:00 2001 From: Hazem Nureldin Date: Tue, 4 Aug 2026 11:43:17 +0100 Subject: [PATCH 03/24] refactor: the log handling on on the webpage --- .../lib/views/BaseSingleWorkflowView.tsx | 104 ++++++++-- .../lib/views/SingleWorkflowView.tsx | 11 +- .../lib/views/TaskLogViewer.tsx | 194 ++++++++++-------- 3 files changed, 207 insertions(+), 102 deletions(-) diff --git a/frontend/relay-workflows-lib/lib/views/BaseSingleWorkflowView.tsx b/frontend/relay-workflows-lib/lib/views/BaseSingleWorkflowView.tsx index 01c6d9aa0..adc586271 100644 --- a/frontend/relay-workflows-lib/lib/views/BaseSingleWorkflowView.tsx +++ b/frontend/relay-workflows-lib/lib/views/BaseSingleWorkflowView.tsx @@ -3,7 +3,6 @@ import { Box, ToggleButton } from "@mui/material"; import { Artifact, Task, - TaskNode, TaskInfo, buildTaskTree, } from "workflows-lib"; @@ -17,6 +16,7 @@ import { BaseSingleWorkflowViewFragment$key } from "./__generated__/BaseSingleWo import BaseWorkflowRelay from "../components/BaseWorkflowRelay"; import { TaskLogViewer } from "./TaskLogViewer"; + export const BaseSingleWorkflowViewFragment = graphql` fragment BaseSingleWorkflowViewFragment on Workflow @relay(mask: false) { name @@ -35,14 +35,18 @@ export const BaseSingleWorkflowViewFragment = graphql` } `; + interface BaseSingleWorkflowViewProps { fragmentRef: BaseSingleWorkflowViewFragment$key | null; taskIds?: string[]; + onSelectTask?: (taskId: string) => void; } + export default function BaseSingleWorkflowView({ taskIds, fragmentRef, + onSelectTask, }: BaseSingleWorkflowViewProps) { const data = useFragment( @@ -50,24 +54,46 @@ export default function BaseSingleWorkflowView({ fragmentRef ); + const fetchedTasks = useFetchedTasks(data ?? null); + + // Task selected for log viewer const [ - selectedTaskIds, - setSelectedTaskIds, - ] = useSelectedTaskIds(); + selectedTaskId, + setSelectedTaskId, + ] = useState(null); const [ filledTaskId, setFilledTaskId, ] = useState(null); + // Resolve task name from id + const selectedTask = useMemo( + () => + fetchedTasks.find( + (task) => task.id === selectedTaskId + ), + [ + fetchedTasks, + selectedTaskId, + ], + ); + - // The task currently opened in the log viewer const [ - selectedTaskId, - setSelectedTaskId, - ] = useState(null); + selectedTaskIds, + setSelectedTaskIds, + ] = useSelectedTaskIds(); + + + // // Artifact hover highlight + // const [ + // filledTaskId, + // setFilledTaskId, + // ] = useState(null); + const taskTree = useMemo( @@ -76,14 +102,19 @@ export default function BaseSingleWorkflowView({ ); + const outputTaskIds: string[] = useMemo(() => { + const newOutputTaskIds: string[] = []; - const traverse = (tasks: TaskNode[]) => { + + const traverse = (tasks: any[]) => { + const sortedTasks = [...tasks].sort( (a, b) => a.id.localeCompare(b.id) ); + sortedTasks.forEach((taskNode) => { if ( @@ -102,8 +133,10 @@ export default function BaseSingleWorkflowView({ } }); + }; + traverse(taskTree); return newOutputTaskIds; @@ -111,31 +144,40 @@ export default function BaseSingleWorkflowView({ }, [taskTree]); + + const handleSelectOutput = () => { setSelectedTaskIds(outputTaskIds); }; + const handleSelectClear = () => { setSelectedTaskIds([]); setSelectedTaskId(null); }; + const onArtifactHover = useCallback( (artifact: Artifact | null) => { + setFilledTaskId( artifact ? artifact.parentTaskId : null ); + }, [] ); + useEffect(() => { + setSelectedTaskIds(taskIds ?? []); + }, [ taskIds, setSelectedTaskIds, @@ -143,8 +185,10 @@ export default function BaseSingleWorkflowView({ + const artifactList: Artifact[] = useMemo(() => { + const filteredTasks = selectedTaskIds.length ? selectedTaskIds @@ -161,10 +205,12 @@ export default function BaseSingleWorkflowView({ : fetchedTasks; + return filteredTasks.flatMap( (task) => task.artifacts ); + }, [ selectedTaskIds, fetchedTasks, @@ -172,12 +218,14 @@ export default function BaseSingleWorkflowView({ + if (!data || !data.status) { return null; } + return ( <> @@ -201,6 +249,7 @@ export default function BaseSingleWorkflowView({ }} > + + + { + + console.log( + "BASE SINGLE TASK SELECTED:", + taskId + ); + + + setSelectedTaskId(taskId); + + + onSelectTask?.(taskId); + + }} /> + + + + + {taskIds && ( */} + + + + ); } \ No newline at end of file diff --git a/frontend/relay-workflows-lib/lib/views/SingleWorkflowView.tsx b/frontend/relay-workflows-lib/lib/views/SingleWorkflowView.tsx index c30e7982b..3dcb3b9fc 100644 --- a/frontend/relay-workflows-lib/lib/views/SingleWorkflowView.tsx +++ b/frontend/relay-workflows-lib/lib/views/SingleWorkflowView.tsx @@ -47,11 +47,12 @@ export default function SingleWorkflowView(props: SingleWorkflowViewProps) { }; return finished || isNull ? ( - - ) : ( + + ) : ( = ({ visit, workflowName, selectedTaskId, + selectedTaskName, }) => { const [logLines, setLogLines] = useState([]); + const [taskCompleted, setTaskCompleted] = useState(false); + const [podName, setPodName] = useState(null); const containerRef = useRef(null); useEffect(() => { console.log("TaskLogViewer selection changed:", { - workflowName, - selectedTaskId, - visit, + workflowName, + selectedTaskId, + visit, }); setLogLines([]); + setTaskCompleted(false); - }, [ + }, [ selectedTaskId, workflowName, visit, - ]); + ]); - - const subscriptionConfig = + const subscriptionConfig = useMemo>( - () => ({ + () => ({ subscription: taskLogViewerSubscription, variables: { - visit, - workflowName, - taskId: selectedTaskId ?? "", + visit, + workflowName, + taskId: selectedTaskId ?? "__NO_TASK_SELECTED__", }, onNext: (payload) => { - console.log("LOG EVENT:", payload); + console.log("LOG EVENT:", payload); - const line = payload?.logs?.content; + const line = payload?.logs?.content; - if (line) { + if (line) { setLogLines((prev) => [ - ...prev, - line, + ...prev, + line, ]); + + if ( + line.includes("sub-process exited") || + line.includes("completed") || + line.includes("finished") || + line.includes("done") + ) { + setTaskCompleted(true); } + } }, onError: (error) => { - console.error("Log subscription error:", error); + console.error("Log subscription error:", error); }, - }), - [ + }), + [ visit, workflowName, selectedTaskId, - ], + ], ); - console.log("SUBSCRIBING WITH:", subscriptionConfig.variables); - // IMPORTANT: - // This hook must ALWAYS run - useSubscription(subscriptionConfig); + console.log( + "SUBSCRIBING WITH:", + subscriptionConfig.variables + ); + // MUST ALWAYS RUN - never put hooks inside conditions + useSubscription(subscriptionConfig); - useEffect(() => { + useEffect(() => { if (containerRef.current) { containerRef.current.scrollTop = containerRef.current.scrollHeight; } - }, [logLines]); return ( - - + } + > + - - - {selectedTaskId ?? "No task selected"} + Logs: {selectedTaskName ?? selectedTaskId ?? "No task selected"} - - {selectedTaskId && ( - + /> )} - + {taskCompleted && ( + + COMPLETED + + )} + - - - {!selectedTaskId && ( - - Select a task. - - )} + > + - {selectedTaskId && - logLines.length === 0 && ( + {!selectedTaskId && ( - Waiting for log output... + Select a task. - )} + )} - {logLines.map((line,index)=>( - - {line} - - ))} + {selectedTaskId && + logLines.length === 0 && ( + + Waiting for log output... + + )} - - - ); + {logLines.map((line, index) => ( + + {line} + + ))} + + + + + + +); }; From c835d836d6368e348d8e0eb8740c086818b9aaba Mon Sep 17 00:00:00 2001 From: Hazem Nureldin Date: Fri, 7 Aug 2026 13:15:46 +0100 Subject: [PATCH 04/24] Support archived S3 logs in subscriptions --- .../graph-proxy/src/graphql/subscription.rs | 339 ++++++++++++++++-- .../lib/components/RelayEnvironment.ts | 8 +- .../lib/views/TaskLogViewer.tsx | 18 +- 3 files changed, 323 insertions(+), 42 deletions(-) diff --git a/backend/graph-proxy/src/graphql/subscription.rs b/backend/graph-proxy/src/graphql/subscription.rs index a3dcea797..6316fa59a 100644 --- a/backend/graph-proxy/src/graphql/subscription.rs +++ b/backend/graph-proxy/src/graphql/subscription.rs @@ -12,6 +12,7 @@ use crate::{ workflows::{Workflow, WorkflowParsingError}, VisitInput, }, + s3client::{Client as S3Client, S3Bucket}, validate_token::ValidatedAuthToken, ArgoServerUrl, }; @@ -46,7 +47,7 @@ struct LogContent { pod_name: String, } -/// Succees/fail events from Workflows API +/// Success/fail events from Workflows API #[derive(Debug, Deserialize)] struct WatchEvent { /// Successful event @@ -58,6 +59,7 @@ struct WatchEvent { /// Get authentication token pub fn get_auth_token(ctx: &Context<'_>) -> anyhow::Result { let auth_token = ctx.data_unchecked::().as_token(); + auth_token .as_ref() .map(|auth| auth.token().to_string()) @@ -66,7 +68,11 @@ pub fn get_auth_token(ctx: &Context<'_>) -> anyhow::Result { #[Subscription(guard = "AuthGuard")] impl WorkflowsSubscription { - /// Processing to subscribe to logs for a single pod of a workflow + /// Subscribe to logs for a single pod of a workflow. + /// + /// Logs are streamed live from Argo while the pod is running. + /// Once the Argo stream finishes, the archived main.log is retrieved + /// from S3 and any lines not already sent are emitted. async fn logs( &self, ctx: &Context<'_>, @@ -76,10 +82,11 @@ impl WorkflowsSubscription { ) -> anyhow::Result>> { let auth_token = get_auth_token(ctx)?; - let namespace = visit.to_string(); let server_url = ctx.data_unchecked::().deref().clone(); let mut url = server_url; + let namespace = visit.to_string(); + url.path_segments_mut().expect("Invalid base URL").extend([ "api", "v1", @@ -95,6 +102,7 @@ impl WorkflowsSubscription { .append_pair("logOptions.follow", "true"); let client = reqwest::Client::new(); + let response = client .get(url) .bearer_auth(auth_token) @@ -104,43 +112,155 @@ impl WorkflowsSubscription { let status = response.status(); let byte_stream = response.bytes_stream(); + + let s3_client = ctx + .data::() + .map_err(|_| anyhow::anyhow!("Missing S3 client"))? + .clone(); + + let s3_bucket = ctx + .data::() + .map_err(|_| anyhow::anyhow!("Missing S3 bucket"))? + .clone(); + + let s3_key = format!("{workflow_name}/{task_id}/main.log"); + let log_stream = stream! { + let mut live_lines = Vec::new(); + for await chunk_result in byte_stream { match chunk_result { Ok(chunk) if status.is_success() => { let text = String::from_utf8_lossy(&chunk).to_string(); + for line in text.lines() { match serde_json::from_str::(line) { Ok(parsed) => { if let Some(result) = parsed.result { + live_lines.push(result.content.clone()); + yield Ok(LogEntry { content: result.content, pod_name: result.pod_name, }); } else { - yield Err("Missing result in log response".to_string()); + yield Err( + "Missing result in log response".to_string() + ); } } + Err(_) => { - yield Ok(LogEntry { - content: line.trim().to_string(), - pod_name: task_id.clone(), - }); + let content = line.trim().to_string(); + + if !content.is_empty() { + live_lines.push(content.clone()); + + yield Ok(LogEntry { + content, + pod_name: task_id.clone(), + }); + } } } } } - Ok(_) | Err(_) => { - yield Err("Failed to read log chunk".to_string()); + + Ok(_) => { + yield Err(format!( + "Argo log request failed with status {status}" + )); + return; + } + + Err(err) => { + yield Err(format!("Failed to read log chunk: {err}")); + return; + } + } + } + + // The live Argo stream has finished. The durable log should now + // be available in the S3 artifact. + let archive_response = match s3_client + .get_object() + .bucket(s3_bucket) + .key(&s3_key) + .send() + .await + { + Ok(response) => response, + + Err(err) => { + yield Err(format!( + "Failed to retrieve archived log artifact: {err:?}" + )); + return; + } + }; + + let archive_bytes = match archive_response.body.collect().await { + Ok(bytes) => bytes, + + Err(err) => { + yield Err(format!( + "Failed to read archived log artifact: {err}" + )); + return; + } + }; + + let archived_text = + String::from_utf8_lossy(archive_bytes.into_bytes().as_ref()).to_string(); + + let archived_lines: Vec = archived_text + .lines() + .map(str::to_string) + .collect(); + + // Determine where the archived log begins beyond what was already + // sent by the live Argo stream. + let mut archive_start = 0; + + while archive_start < live_lines.len() + && archive_start < archived_lines.len() + && live_lines[archive_start] == archived_lines[archive_start] + { + archive_start += 1; + } + + // If the archive and live stream don't share the same prefix, + // try to locate the final live line in the archive. + if archive_start < live_lines.len() { + if let Some(last_live_line) = live_lines.last() { + if let Some(position) = archived_lines + .iter() + .rposition(|line| line == last_live_line) + { + archive_start = position + 1; + } else { + yield Err( + "Unable to reconcile live and archived logs".to_string() + ); + return; } } } + + // Send only the archived lines that were not already emitted + // from the live Argo stream. + for line in archived_lines.into_iter().skip(archive_start) { + yield Ok(LogEntry { + content: line, + pod_name: task_id.clone(), + }); + } }; Ok(log_stream) } - /// Processing to subscribe to data for all workflows in a session + /// Subscribe to data for all workflows in a session. async fn workflow( &self, ctx: &Context<'_>, @@ -166,6 +286,7 @@ impl WorkflowsSubscription { ); let client = reqwest::Client::new(); + let response = client .get(url) .bearer_auth(auth_token) @@ -177,6 +298,7 @@ impl WorkflowsSubscription { let stream = response.then(move |event_result| { let session_clone = visit.clone(); + async move { match event_result { Ok(event) => { @@ -191,14 +313,18 @@ impl WorkflowsSubscription { Err("No workflow object returned".to_string()) } } + (None, Some(err)) => Err(err.message), + (None, None) => Err("Missing result and error in event".to_string()), + (Some(_), Some(_)) => { Err("Conflicting result and error in event".to_string()) } } } - Err(_err) => Err("Failed to read event from stream".to_string()), + + Err(_) => Err("Failed to read event from stream".to_string()), } } }); @@ -216,7 +342,6 @@ struct StreamError { #[cfg(test)] mod tests { - use std::{env, fs, path::PathBuf}; use async_graphql::Request; @@ -227,20 +352,157 @@ mod tests { use serde_json::{json, Value}; use url::Url; - use crate::graphql::Visit; - use crate::ArgoServerUrl; - use crate::graphql::root_schema_builder; + use crate::graphql::Visit; use crate::validate_token::ValidatedAuthToken; + use crate::{ArgoServerUrl, Client, S3Bucket, S3ClientArgs}; fn test_token() -> ValidatedAuthToken { let token = Authorization::bearer("test-token").expect("token always valid"); + ValidatedAuthToken::Valid(token) } + #[tokio::test] + async fn logs_subscription_reads_archived_s3_log_after_live_stream() { + let workflow_name = "numpy-benchmark-wdkwj"; + let task_id = "numpy-benchmark-wdkwj"; + + let visit = Visit { + proposal_code: "mg".to_string(), + proposal_number: 36964, + number: 1, + }; + + let mut server = mockito::Server::new_async().await; + + // Mock the live Argo log endpoint. + let argo_log_body = concat!( + r#"{"result":{"content":"line 1","podName":"numpy-benchmark-wdkwj"}}"#, + "\n", + r#"{"result":{"content":"line 2","podName":"numpy-benchmark-wdkwj"}}"#, + "\n", + ); + + let argo_log_path = format!("/api/v1/workflows/{visit}/{workflow_name}/log"); + + let argo_log_endpoint = server + .mock("GET", argo_log_path.as_str()) + .match_query(Matcher::UrlEncoded("podName".into(), task_id.into())) + .match_query(Matcher::UrlEncoded( + "logOptions.container".into(), + "main".into(), + )) + .match_query(Matcher::UrlEncoded( + "logOptions.follow".into(), + "true".into(), + )) + .with_status(200) + .with_header("content-type", "text/plain") + .with_body(argo_log_body) + .create_async() + .await; + + // Mock the archived S3 main.log. + // + // Path-style S3 addressing produces: + // + // /test-bucket/numpy-benchmark-wdkwj/numpy-benchmark-wdkwj/main.log + let s3_key = format!("{workflow_name}/{task_id}/main.log"); + // let s3_path = format!("/test-bucket/{s3_key}"); + + let s3_log_endpoint = server + .mock("GET", mockito::Matcher::Any) + .with_status(200) + .with_header("content-type", "text/plain") + .with_body("line 1\nline 2\nline 3\nline 4\n") + .create_async() + .await; + + let s3_bucket = S3Bucket("test-bucket".to_string()); + + let s3_client_args = S3ClientArgs { + s3_endpoint_url: Some(Url::parse(&server.url()).unwrap()), + s3_access_key_id: Some("test-access-key".to_string()), + s3_secret_access_key: Some("test-secret-key".to_string()), + s3_force_path_style: true, + s3_region: Some("us-west-2".to_string()), + }; + + let s3_client = Client::from(s3_client_args); + + let argo_server_url = Url::parse(&server.url()).unwrap(); + + let schema = root_schema_builder() + .data(ArgoServerUrl(argo_server_url)) + .data(test_token()) + .data(s3_client) + .data(s3_bucket) + .finish(); + + let request = Request::new(format!( + r#" + subscription {{ + logs( + visit: {{ + proposalCode: "{}", + proposalNumber: {}, + number: {} + }} + workflowName: "{}" + taskId: "{}" + ) {{ + content + podName + }} + }} + "#, + visit.proposal_code, visit.proposal_number, visit.number, workflow_name, task_id, + )); + + let mut response_stream = schema.execute_stream(request); + + let mut logs = Vec::new(); + + while let Some(response) = response_stream.next().await { + assert!( + response.errors.is_empty(), + "unexpected GraphQL errors: {:?}", + response.errors + ); + + let data = response.data.into_json().expect("invalid response JSON"); + + if let Some(log) = data.get("logs").and_then(|value| value.as_object()) { + logs.push(( + log["content"].as_str().unwrap().to_string(), + log["podName"].as_str().unwrap().to_string(), + )); + } + + if logs.len() == 4 { + break; + } + } + + assert_eq!( + logs, + vec![ + ("line 1".to_string(), task_id.to_string()), + ("line 2".to_string(), task_id.to_string()), + ("line 3".to_string(), task_id.to_string()), + ("line 4".to_string(), task_id.to_string()), + ] + ); + + argo_log_endpoint.assert_async().await; + s3_log_endpoint.assert_async().await; + } + #[tokio::test] async fn single_workflow_subscription_returns_first_event() { let workflow_name = "numpy-benchmark-wdkwj"; + let visit = Visit { proposal_code: "mg".to_string(), proposal_number: 36964, @@ -248,6 +510,7 @@ mod tests { }; let mut workflow_file_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")); + workflow_file_path.push("test-assets"); workflow_file_path.push("get-workflow-wdkwj.json"); @@ -270,7 +533,9 @@ mod tests { ); let mut server = mockito::Server::new_async().await; + let path = format!("/api/v1/workflow-events/{visit}"); + let workflow_events_endpoint = server .mock("GET", path.as_str()) .match_query(Matcher::UrlEncoded( @@ -292,15 +557,19 @@ mod tests { let request = Request::new(format!( r#" - subscription {{ - workflow( - name: "{}", - visit: {{ proposalCode: "{}", proposalNumber: {}, number: {} }} - ) {{ - name + subscription {{ + workflow( + name: "{}", + visit: {{ + proposalCode: "{}", + proposalNumber: {}, + number: {} + }} + ) {{ + name + }} }} - }} - "#, + "#, workflow_name, visit.proposal_code, visit.proposal_number, visit.number )); @@ -346,15 +615,19 @@ mod tests { let request = Request::new( r#" - subscription { - workflow( - name: "workflowName", - visit: { proposalCode: "xy", proposalNumber: 1234, number: 5678 } - ) { - name + subscription { + workflow( + name: "workflowName", + visit: { + proposalCode: "xy", + proposalNumber: 1234, + number: 5678 + } + ) { + name + } } - } - "#, + "#, ); let mut response_stream = schema.execute_stream(request); @@ -365,6 +638,7 @@ mod tests { .expect("subscription stream ended before first response"); let expected_data = json!(null); + assert_eq!( first_response .data @@ -384,6 +658,7 @@ mod tests { .expect("invalid json"); let expected_value = json!(AuthErrorCode::Unauthenticated.to_string()); + assert_eq!(error_code, expected_value); } } diff --git a/frontend/relay-workflows-lib/lib/components/RelayEnvironment.ts b/frontend/relay-workflows-lib/lib/components/RelayEnvironment.ts index 571bca628..5466734f8 100644 --- a/frontend/relay-workflows-lib/lib/components/RelayEnvironment.ts +++ b/frontend/relay-workflows-lib/lib/components/RelayEnvironment.ts @@ -96,10 +96,10 @@ console.log("WS_ENDPOINTYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYY:", WS_ENDPOI export const wsClient = createClient({ url: WS_ENDPOINT, on: { - connecting: () => console.log("WS connecting"), - opened: () => console.log("WS opened"), - connected: () => console.log("WS connected"), - closed: (event) => console.log("WS closed", event), + connecting: () => { console.log("WS connecting"); }, + opened: () => { console.log("WS opened"); }, + connected: () => { console.log("WS connected"); }, + closed: (event) => { console.log("WS closed", event); }, }, webSocketImpl: class extends WebSocket { constructor(url: string | URL, protocols?: string | string[]) { diff --git a/frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx b/frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx index 80a8300c9..96169eb4f 100644 --- a/frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx +++ b/frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx @@ -59,6 +59,8 @@ export const TaskLogViewer: React.FC = ({ const containerRef = useRef(null); + const [expanded, setExpanded] = useState(false); + useEffect(() => { console.log("TaskLogViewer selection changed:", { workflowName, @@ -68,6 +70,7 @@ export const TaskLogViewer: React.FC = ({ setLogLines([]); setTaskCompleted(false); + setExpanded(!!selectedTaskId); }, [ selectedTaskId, @@ -142,11 +145,13 @@ export const TaskLogViewer: React.FC = ({ return ( { setExpanded(isExpanded); }} sx={{ - mt: 2, - width: "100%", - backgroundColor: "#001400", - color: "#00ff00", + mt: 2, + width: "100%", + backgroundColor: "#001400", + color: "#00ff00", }} > @@ -167,9 +172,10 @@ export const TaskLogViewer: React.FC = ({ {selectedTaskId && !taskCompleted && ( )} From 12bba891bfe4242fea7cff26279a71726baf58f5 Mon Sep 17 00:00:00 2001 From: Hazem Nureldin Date: Sun, 16 Aug 2026 23:01:26 +0100 Subject: [PATCH 05/24] fix: display logs like the archieved ones --- .../graph-proxy/src/graphql/subscription.rs | 31 +++++++++++++++++-- 1 file changed, 29 insertions(+), 2 deletions(-) diff --git a/backend/graph-proxy/src/graphql/subscription.rs b/backend/graph-proxy/src/graphql/subscription.rs index 6316fa59a..af77986fc 100644 --- a/backend/graph-proxy/src/graphql/subscription.rs +++ b/backend/graph-proxy/src/graphql/subscription.rs @@ -101,6 +101,14 @@ impl WorkflowsSubscription { .append_pair("logOptions.container", "main") .append_pair("logOptions.follow", "true"); + + tracing::info!( + "LOG REQUEST namespace={} workflow={} task={}", + namespace, + workflow_name, + task_id + ); + let client = reqwest::Client::new(); let response = client @@ -137,10 +145,23 @@ impl WorkflowsSubscription { match serde_json::from_str::(line) { Ok(parsed) => { if let Some(result) = parsed.result { - live_lines.push(result.content.clone()); + let content = result.content; + + let skip_line = + content.contains("capturing logs") + || content.contains("waiting for signals") + || content.contains("sub-process exited") + || content.contains("file signal handler exiting") + || content.contains("no need to save artifact"); + + if skip_line { + continue; + } + + live_lines.push(content.clone()); yield Ok(LogEntry { - content: result.content, + content, pod_name: result.pod_name, }); } else { @@ -153,6 +174,10 @@ impl WorkflowsSubscription { Err(_) => { let content = line.trim().to_string(); + if content.starts_with("{\"result\"") { + continue; + } + if !content.is_empty() { live_lines.push(content.clone()); @@ -162,6 +187,8 @@ impl WorkflowsSubscription { }); } } + + } } } From 62f7e9e05d75d731689e6e2fe257a18ad310057a Mon Sep 17 00:00:00 2001 From: Hazem Nureldin Date: Mon, 17 Aug 2026 10:08:44 +0100 Subject: [PATCH 06/24] fix: the link to s3 main.log working --- backend/graph-proxy/src/graphql/subscription.rs | 7 +++---- backend/graph-proxy/src/graphql/workflows.rs | 11 ++++++----- 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/backend/graph-proxy/src/graphql/subscription.rs b/backend/graph-proxy/src/graphql/subscription.rs index af77986fc..8ae65e13d 100644 --- a/backend/graph-proxy/src/graphql/subscription.rs +++ b/backend/graph-proxy/src/graphql/subscription.rs @@ -101,13 +101,12 @@ impl WorkflowsSubscription { .append_pair("logOptions.container", "main") .append_pair("logOptions.follow", "true"); - tracing::info!( "LOG REQUEST namespace={} workflow={} task={}", namespace, workflow_name, task_id - ); + ); let client = reqwest::Client::new(); @@ -435,8 +434,8 @@ mod tests { // Path-style S3 addressing produces: // // /test-bucket/numpy-benchmark-wdkwj/numpy-benchmark-wdkwj/main.log - let s3_key = format!("{workflow_name}/{task_id}/main.log"); - // let s3_path = format!("/test-bucket/{s3_key}"); + let _s3_key = format!("{workflow_name}/{task_id}/main.log"); + // let s3_path = format!("/test-bucket/{s3_key}"); let s3_log_endpoint = server .mock("GET", mockito::Matcher::Any) diff --git a/backend/graph-proxy/src/graphql/workflows.rs b/backend/graph-proxy/src/graphql/workflows.rs index e40240d8f..488a6a115 100644 --- a/backend/graph-proxy/src/graphql/workflows.rs +++ b/backend/graph-proxy/src/graphql/workflows.rs @@ -373,16 +373,17 @@ impl Artifact<'_> { .expires_in(std::time::Duration::from_secs(3600)) .build() .unwrap(); - s3_client + let req = s3_client .get_object() .bucket(s3_bucket.clone()) .key(key) .presigned(presigning_config) .await - .map_err(|_| WorkflowParsingError::InvalidPresignedS3Url) - .and_then(|req| { - Url::parse(req.uri()).map_err(|_| WorkflowParsingError::InvalidPresignedS3Url) - }) + .map_err(|_| WorkflowParsingError::InvalidPresignedS3Url)?; + + tracing::info!("PRESIGNED URL: {}", req.uri()); + + Url::parse(req.uri()).map_err(|_| WorkflowParsingError::InvalidPresignedS3Url) } /// The MIME type of the artifact data From b74ed4e1f6eb68b088b9f2aa2ebc45a50c78154d Mon Sep 17 00:00:00 2001 From: Hazem Nureldin Date: Tue, 18 Aug 2026 11:55:47 +0100 Subject: [PATCH 07/24] fix:to be able to retreive old logs of the workflows --- .../graph-proxy/src/graphql/subscription.rs | 133 ++++----- .../lib/views/TaskLogViewer.tsx | 265 ++++++++++-------- 2 files changed, 215 insertions(+), 183 deletions(-) diff --git a/backend/graph-proxy/src/graphql/subscription.rs b/backend/graph-proxy/src/graphql/subscription.rs index 8ae65e13d..bb47304ba 100644 --- a/backend/graph-proxy/src/graphql/subscription.rs +++ b/backend/graph-proxy/src/graphql/subscription.rs @@ -110,15 +110,13 @@ impl WorkflowsSubscription { let client = reqwest::Client::new(); - let response = client + // Try Argo, but don't fail the whole subscription if it errors. + let argo_response = client .get(url) .bearer_auth(auth_token) .header("Accept", "text/plain") .send() - .await?; - - let status = response.status(); - let byte_stream = response.bytes_stream(); + .await; let s3_client = ctx .data::() @@ -135,79 +133,88 @@ impl WorkflowsSubscription { let log_stream = stream! { let mut live_lines = Vec::new(); - for await chunk_result in byte_stream { - match chunk_result { - Ok(chunk) if status.is_success() => { - let text = String::from_utf8_lossy(&chunk).to_string(); - - for line in text.lines() { - match serde_json::from_str::(line) { - Ok(parsed) => { - if let Some(result) = parsed.result { - let content = result.content; - - let skip_line = - content.contains("capturing logs") - || content.contains("waiting for signals") - || content.contains("sub-process exited") - || content.contains("file signal handler exiting") - || content.contains("no need to save artifact"); - - if skip_line { - continue; + // --- Live Argo stream (optional) --- + if let Ok(response) = argo_response { + let status = response.status(); + let byte_stream = response.bytes_stream(); + + for await chunk_result in byte_stream { + match chunk_result { + Ok(chunk) if status.is_success() => { + let text = String::from_utf8_lossy(&chunk).to_string(); + + for line in text.lines() { + match serde_json::from_str::(line) { + Ok(parsed) => { + if let Some(result) = parsed.result { + let content = result.content; + + let skip_line = + content.contains("capturing logs") + || content.contains("waiting for signals") + || content.contains("sub-process exited") + || content.contains("file signal handler exiting") + || content.contains("no need to save artifact") + || content.contains("no need to save parameter"); + + if skip_line { + continue; + } + + live_lines.push(content.clone()); + + yield Ok(LogEntry { + content, + pod_name: result.pod_name, + }); + } else { + yield Err( + "Missing result in log response".to_string() + ); } - - live_lines.push(content.clone()); - - yield Ok(LogEntry { - content, - pod_name: result.pod_name, - }); - } else { - yield Err( - "Missing result in log response".to_string() - ); } - } - Err(_) => { - let content = line.trim().to_string(); + Err(_) => { + let content = line.trim().to_string(); - if content.starts_with("{\"result\"") { - continue; - } + if content.starts_with("{\"result\"") { + continue; + } - if !content.is_empty() { - live_lines.push(content.clone()); + if !content.is_empty() { + live_lines.push(content.clone()); - yield Ok(LogEntry { - content, - pod_name: task_id.clone(), - }); + yield Ok(LogEntry { + content, + pod_name: task_id.clone(), + }); + } } } - - } } - } - Ok(_) => { - yield Err(format!( - "Argo log request failed with status {status}" - )); - return; - } + Ok(_) => { + // Argo failed (e.g. 404), log and continue to S3. + tracing::warn!( + "Argo log request failed with status {status}, will try S3 fallback" + ); + } - Err(err) => { - yield Err(format!("Failed to read log chunk: {err}")); - return; + Err(err) => { + tracing::warn!( + "Failed to read log chunk from Argo: {err}, will try S3 fallback" + ); + } } } + } else { + tracing::warn!("Argo log request failed entirely, will try S3 fallback"); } - // The live Argo stream has finished. The durable log should now - // be available in the S3 artifact. + // --- S3 fallback (always attempted) --- + tracing::info!("ARCHIVE_LOOKUP: {}", s3_key); + let archive_response = match s3_client .get_object() .bucket(s3_bucket) @@ -286,7 +293,7 @@ impl WorkflowsSubscription { Ok(log_stream) } - /// Subscribe to data for all workflows in a session. +/// Subscribe to data for all workflows in a session. async fn workflow( &self, ctx: &Context<'_>, diff --git a/frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx b/frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx index 96169eb4f..a8643f78c 100644 --- a/frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx +++ b/frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx @@ -18,7 +18,6 @@ import { GraphQLSubscriptionConfig } from "relay-runtime"; import { Visit } from "@diamondlightsource/sci-react-ui"; import { TaskLogViewerSubscription } from "./__generated__/TaskLogViewerSubscription.graphql"; - const taskLogViewerSubscription = graphql` subscription TaskLogViewerSubscription( $visit: VisitInput! @@ -36,7 +35,6 @@ const taskLogViewerSubscription = graphql` } `; - interface TaskLogViewerProps { visit: Visit; workflowName: string; @@ -44,168 +42,213 @@ interface TaskLogViewerProps { selectedTaskName?: string; } - export const TaskLogViewer: React.FC = ({ visit, workflowName, selectedTaskId, selectedTaskName, }) => { + console.log("TASK VIEWER PROPS", { + selectedTaskId, + selectedTaskName, + workflowName, + }); + + // --- ALL HOOKS MUST BE AT THE TOP, BEFORE ANY EARLY RETURN --- const [logLines, setLogLines] = useState([]); const [taskCompleted, setTaskCompleted] = useState(false); const [podName, setPodName] = useState(null); const containerRef = useRef(null); - - const [expanded, setExpanded] = useState(false); useEffect(() => { - console.log("TaskLogViewer selection changed:", { - workflowName, - selectedTaskId, - visit, + console.log("CLEARING LOGS - TaskLogViewer selection changed:", { + workflowName, + selectedTaskId, + visit, }); setLogLines([]); setTaskCompleted(false); setExpanded(!!selectedTaskId); + }, [selectedTaskId, workflowName, visit]); - }, [ + console.log("SELECTED TASK:", { selectedTaskId, - workflowName, - visit, - ]); + selectedTaskName, + }); + // Build subscription config unconditionally + const subscriptionConfig = useMemo>( + () => ({ + subscription: taskLogViewerSubscription, - const subscriptionConfig = - useMemo>( - () => ({ - subscription: taskLogViewerSubscription, + variables: { + visit, + workflowName, + taskId: selectedTaskId ?? "", + }, - variables: { - visit, - workflowName, - taskId: selectedTaskId ?? "__NO_TASK_SELECTED__", - }, + skip: !selectedTaskId, - onNext: (payload) => { - console.log("LOG EVENT:", payload); + onNext: (payload) => { + console.log("LOG RECEIVED:", payload); const line = payload?.logs?.content; if (line) { - setLogLines((prev) => [ - ...prev, - line, - ]); + console.log("APPENDING:", line); + setLogLines((prev) => [...prev, line]); - if ( + if ( line.includes("sub-process exited") || line.includes("completed") || line.includes("finished") || + line.includes("Image saved") || + line.includes("saved image") || line.includes("done") - ) { + ) { setTaskCompleted(true); - } + } } - }, - - onError: (error) => { - console.error("Log subscription error:", error); - }, - }), - [ - visit, - workflowName, - selectedTaskId, - ], - ); - - - console.log( - "SUBSCRIBING WITH:", - subscriptionConfig.variables + }, + + onCompleted: () => { + console.log("LOG SUBSCRIPTION COMPLETED"); + setTaskCompleted(true); + }, + + onError: (error) => { + console.error("Log subscription error:", error); + // Treat errors as "done" for UI so we don't show "Waiting..." forever + setTaskCompleted(true); + }, + }), + [visit, workflowName, selectedTaskId] ); + console.log("SUBSCRIBING WITH:", subscriptionConfig.variables); + console.log("TASK VIEWER", { + selectedTaskId, + skip: !selectedTaskId, + }); - - // MUST ALWAYS RUN - never put hooks inside conditions + // ALWAYS called, regardless of selectedTaskId useSubscription(subscriptionConfig); - useEffect(() => { if (containerRef.current) { - containerRef.current.scrollTop = - containerRef.current.scrollHeight; + containerRef.current.scrollTop = containerRef.current.scrollHeight; } }, [logLines]); + // --- NOW YOU CAN DO EARLY RETURN FOR UI ONLY --- - - return ( - { setExpanded(isExpanded); }} + if (!selectedTaskId) { + return ( + - + > - } + } > + + Logs: No task selected + + + + + Select a task. + + + + ); + } + + // Main UI when a task IS selected + return ( + setExpanded(isExpanded)} + sx={{ + mt: 2, + width: "100%", + backgroundColor: "#001400", + color: "#00ff00", + }} + > + } + > - Logs: {selectedTaskName ?? selectedTaskId ?? "No task selected"} + Logs: {selectedTaskName ?? selectedTaskId} - {selectedTaskId && !taskCompleted && ( - + /> )} {taskCompleted && ( - + > COMPLETED - + )} + - - - - - + = ({ fontFamily: "monospace", fontSize: "10px", whiteSpace: "pre-wrap", - }} + }} > + {selectedTaskId && logLines.length === 0 && !taskCompleted && ( + Waiting for log output... + )} - {!selectedTaskId && ( - - Select a task. - - )} - - - {selectedTaskId && - logLines.length === 0 && ( - - Waiting for log output... - - )} - - - {logLines.map((line, index) => ( - - {line} - - ))} - + {logLines.map((line, index) => ( + {line} + ))} - - - + -); + ); }; - export default TaskLogViewer; \ No newline at end of file From af747508bbcfcf53a70778af63ae84cde929ce41 Mon Sep 17 00:00:00 2001 From: Hazem Nureldin Date: Tue, 18 Aug 2026 14:13:15 +0100 Subject: [PATCH 08/24] style: selct type of files to show in output --- .../workflows-lib/lib/components/workflow/TaskInfo.tsx | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/frontend/workflows-lib/lib/components/workflow/TaskInfo.tsx b/frontend/workflows-lib/lib/components/workflow/TaskInfo.tsx index 5becae078..b42c18c00 100644 --- a/frontend/workflows-lib/lib/components/workflow/TaskInfo.tsx +++ b/frontend/workflows-lib/lib/components/workflow/TaskInfo.tsx @@ -9,7 +9,7 @@ import ArrowDropDownIcon from "@mui/icons-material/ArrowDropDown"; import { ArtifactFilteredList } from "./ArtifactFilteredList"; import type { Artifact } from "workflows-lib"; import { ImageInfo, ScrollableImages } from "./ScrollableImages"; -import { useState, useMemo } from "react"; +import { useState, useMemo, useEffect } from "react"; import { FuzzySearchBar } from "./FuzzySearchBar"; import { FileTypeDropdown } from "./FileTypeDropdown"; import Fuse from "fuse.js"; @@ -37,6 +37,14 @@ const TaskInfo: React.FC = ({ return types; }, [artifactList]); + + // Default: all types except .log + useEffect(() => { + if (fileTypes.length === 0) return; + setSelectedFileTypes(fileTypes.filter((t) => t !== ".log")); + }, [fileTypes]); + + const filteredArtifactList: Artifact[] = useMemo(() => { let filtered = artifactList; From 5c7933451fdf9fba295e384e8285ce62c92d3d6e Mon Sep 17 00:00:00 2001 From: Hazem Nureldin Date: Tue, 18 Aug 2026 14:19:09 +0100 Subject: [PATCH 09/24] style: formating code --- backend/graph-proxy/src/graphql/subscription.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/backend/graph-proxy/src/graphql/subscription.rs b/backend/graph-proxy/src/graphql/subscription.rs index bb47304ba..4956bc3f9 100644 --- a/backend/graph-proxy/src/graphql/subscription.rs +++ b/backend/graph-proxy/src/graphql/subscription.rs @@ -293,7 +293,7 @@ impl WorkflowsSubscription { Ok(log_stream) } -/// Subscribe to data for all workflows in a session. + /// Subscribe to data for all workflows in a session. async fn workflow( &self, ctx: &Context<'_>, From baa427be273df267a89f10e7cd2968380055222f Mon Sep 17 00:00:00 2001 From: Hazem Nureldin Date: Wed, 19 Aug 2026 18:37:02 +0100 Subject: [PATCH 10/24] feat: parameters filter --- backend/graph-proxy/src/graphql/filters.rs | 284 ++++++++++++++++++- backend/graph-proxy/src/graphql/workflows.rs | 24 +- 2 files changed, 289 insertions(+), 19 deletions(-) diff --git a/backend/graph-proxy/src/graphql/filters.rs b/backend/graph-proxy/src/graphql/filters.rs index d63024bbb..84dda2747 100644 --- a/backend/graph-proxy/src/graphql/filters.rs +++ b/backend/graph-proxy/src/graphql/filters.rs @@ -1,5 +1,6 @@ use std::collections::HashSet; +use argo_workflows_openapi::IoArgoprojWorkflowV1alpha1Workflow; use async_graphql::{ Enum, InputObject, InputValueError, InputValueResult, Scalar, ScalarType, Value, }; @@ -114,7 +115,6 @@ pub enum WorkflowLabelSelectorOperator { /// Represents a label selector for filtering workflows based on labels #[derive(Debug, Clone, InputObject)] -/// Represents a label selector for filtering workflows based on labels pub struct LabelSelector { /// The label key to filter on key: String, @@ -126,19 +126,34 @@ pub struct LabelSelector { // Workflows-------------------------------------------- +/// Represents a workflow parameter filter +#[derive(Debug, Clone, InputObject)] +pub struct WorkflowParameterFilter { + /// The workflow parameter name + key: String, + + /// The workflow parameter value + value: String, +} + /// All the supported Workflows filters #[derive(Debug, Default, Clone, InputObject)] pub struct WorkflowFilter { /// The status of the workflow (e.g., pending, running, succeeded, failed, error) workflow_status_filter: Option, + /// The fedid of the user who created the workflow creator: Option, + /// The workflow template template: Option