From cddba6f09864f553cf0db19686525ad866c845db Mon Sep 17 00:00:00 2001 From: Michele Pangrazzi Date: Fri, 28 Aug 2026 09:52:50 +0200 Subject: [PATCH] feat(dashboard): surface durable execution attempts --- PRODUCT.md | 33 ++++++++++++++++ dashboard/src/components/TraceCard.test.tsx | 18 +++++++++ dashboard/src/constants.ts | 18 +++++++++ dashboard/src/index.css | 4 ++ dashboard/src/types.ts | 2 +- dashboard/src/utils/traces.test.ts | 1 + dashboard/src/utils/traces.ts | 1 + src/hayhooks/server/tracing.py | 19 ++++++++- src/hayhooks/server/utils/deploy_utils.py | 43 ++++++++++++++++++++- tests/test_durable_hayhooks.py | 14 ++++++- tests/test_tracing.py | 20 ++++++++++ 11 files changed, 168 insertions(+), 5 deletions(-) create mode 100644 PRODUCT.md diff --git a/PRODUCT.md b/PRODUCT.md new file mode 100644 index 00000000..3524d69e --- /dev/null +++ b/PRODUCT.md @@ -0,0 +1,33 @@ +# Product + +## Register + +product + +## Users + +Hayhooks developers and operators inspecting local live activity while building, testing, or diagnosing deployed Pipelines and Agents. They need to recognize execution state, failures, and latency without leaving the task at hand. + +## Product Purpose + +The dashboard makes Hayhooks runtime behavior legible through a focused live trace view. It should surface durable submissions and attempts alongside existing runtime activity, while remaining an observability tool rather than an execution control plane. + +## Brand Personality + +Precise, calm, utilitarian. The interface should feel trustworthy under debugging pressure and keep attention on the runtime data. + +## Anti-references + +Avoid decorative analytics dashboards, dense enterprise observability consoles, unfamiliar controls, and state that depends on color alone. Do not imply durable A2A support before that lifecycle is implemented. + +## Design Principles + +- Extend the dashboard's existing vocabulary before introducing new patterns. +- Put execution identity and state where they support diagnosis. +- Prefer concise summaries with details available progressively. +- Keep local-buffer and durability boundaries honest in the interface. +- Preserve focus and readability as traces update live. + +## Accessibility & Inclusion + +Target WCAG AA contrast and keyboard access. Keep focus states and semantic labels intact, honor reduced-motion preferences, and communicate status with text or shape in addition to color. diff --git a/dashboard/src/components/TraceCard.test.tsx b/dashboard/src/components/TraceCard.test.tsx index 43de2cdc..3b4f2e77 100644 --- a/dashboard/src/components/TraceCard.test.tsx +++ b/dashboard/src/components/TraceCard.test.tsx @@ -113,6 +113,24 @@ describe("TraceCard", () => { expect(freshCard).not.toHaveClass("trace-card-fresh-run") }) + it("shows durable executions and checkpoint state", () => { + const trace = makeTrace({ + root_span: makeSpan({ name: "hayhooks.durable.attempt" }), + tags: [ + { key: "hayhooks.durable.execution_id", value: "execution-123" }, + { key: "hayhooks.durable.attempt", value: "2" }, + { key: "hayhooks.checkpoint", value: "true" }, + { key: "hayhooks.success", value: "true" }, + ], + }) + render() + + expect(screen.getByText("durable")).toBeInTheDocument() + expect(screen.getByText("execution-123")).toBeInTheDocument() + expect(screen.getByText("checkpoint")).toBeInTheDocument() + expect(screen.queryByText("failed")).not.toBeInTheDocument() + }) + it("shows related tags for the selected span", () => { const childSpan = makeSpan({ span_id: "span-child", diff --git a/dashboard/src/constants.ts b/dashboard/src/constants.ts index d562ef1e..36533add 100644 --- a/dashboard/src/constants.ts +++ b/dashboard/src/constants.ts @@ -13,6 +13,10 @@ export const DEFAULT_DASHBOARD_CONFIG: DashboardConfig = { export const TAG_PRIORITY = [ "hayhooks.pipeline.name", "hayhooks.transport", + "hayhooks.durable.execution_id", + "hayhooks.durable.attempt", + "hayhooks.durable.kind", + "hayhooks.durable.definition_revision", "hayhooks.route", "hayhooks.openai.operation", "hayhooks.openai.stream_requested", @@ -21,6 +25,7 @@ export const TAG_PRIORITY = [ "hayhooks.response.streaming", "hayhooks.payload.values", "hayhooks.payload.has_files", + "hayhooks.checkpoint", "hayhooks.success", "hayhooks.error.type", "hayhooks.http.status_code", @@ -35,6 +40,10 @@ export const TAG_PRIORITY = [ export const TAG_LABELS: Record = { "hayhooks.pipeline.name": "pipeline", "hayhooks.transport": "transport", + "hayhooks.durable.execution_id": "execution", + "hayhooks.durable.attempt": "attempt", + "hayhooks.durable.kind": "durable kind", + "hayhooks.durable.definition_revision": "revision", "hayhooks.openai.operation": "openai op", "hayhooks.openai.stream_requested": "stream", "hayhooks.openai.execution_mode": "exec mode", @@ -50,12 +59,16 @@ export const TAG_LABELS: Record = { "hayhooks.route": "route", "hayhooks.payload.values": "payload values", "hayhooks.payload.has_files": "has files", + "hayhooks.checkpoint": "checkpoint", "service.name": "service", serviceName: "service", } export const SUMMARY_TAG_KEYS = new Set([ "hayhooks.transport", + "hayhooks.durable.execution_id", + "hayhooks.durable.attempt", + "hayhooks.checkpoint", "hayhooks.success", "hayhooks.error.type", ]) @@ -86,6 +99,11 @@ export const KIND_STYLE: Record { ["hayhooks.deploy", "deploy"], ["hayhooks.openai.chat", "openai"], ["hayhooks.mcp.tool", "mcp"], + ["hayhooks.durable.attempt", "durable"], ["hayhooks.run", "run"], ["something.else", "other"], ] as const)("classifies '%s' as '%s'", (spanName, expected) => { diff --git a/dashboard/src/utils/traces.ts b/dashboard/src/utils/traces.ts index 7e471a35..5ab31e6c 100644 --- a/dashboard/src/utils/traces.ts +++ b/dashboard/src/utils/traces.ts @@ -8,6 +8,7 @@ const TRACE_KIND_RULES: Array<{ includes: string; kind: TraceKind }> = [ { includes: ".deploy", kind: "deploy" }, { includes: ".openai.", kind: "openai" }, { includes: ".mcp.", kind: "mcp" }, + { includes: ".durable.", kind: "durable" }, { includes: ".run", kind: "run" }, ] diff --git a/src/hayhooks/server/tracing.py b/src/hayhooks/server/tracing.py index c21c7b17..916231d0 100644 --- a/src/hayhooks/server/tracing.py +++ b/src/hayhooks/server/tracing.py @@ -41,6 +41,7 @@ SPAN_PIPELINE_UNDEPLOY = "hayhooks.pipeline.undeploy" SPAN_PIPELINE_RUN = "hayhooks.pipeline.run" SPAN_PIPELINE_STARTUP_DEPLOY = "hayhooks.pipeline.startup.deploy" +SPAN_DURABLE_ATTEMPT = "hayhooks.durable.attempt" SPAN_OPENAI_RUN = "hayhooks.openai.run" SPAN_OPENAI_FILE_UPLOAD = "hayhooks.openai.file_upload" SPAN_MCP_LIST_TOOLS = "hayhooks.mcp.list_tools" @@ -56,6 +57,7 @@ _TAG_HTTP_STATUS = "hayhooks.http.status_code" _TAG_RESPONSE_STREAMING = "hayhooks.response.streaming" _TAG_RESPONSE_STREAM_TYPE = "hayhooks.response.stream_type" +_TAG_CHECKPOINT = "hayhooks.checkpoint" _FASTAPI_STATE_FLAG = "_hayhooks_fastapi_instrumented" _STARLETTE_STATE_FLAG = "_hayhooks_starlette_instrumented" @@ -122,6 +124,14 @@ def _span_correlation_data(span: Span | None) -> dict[str, str]: return normalize_trace_correlation_data(correlation_data) if correlation_data else {} +def _is_checkpoint_exception(exc: BaseException) -> bool: + exception_type = type(exc) + return (exception_type.__name__ == "BreakpointException" and exception_type.__module__.startswith("haystack.")) or ( + exception_type.__name__ == "_ExecutionSuspendedError" + and exception_type.__module__ == "hayhooks.durable.context" + ) + + def _record_live_span_outcome( *, trace_id: str, @@ -131,8 +141,10 @@ def _record_live_span_outcome( ) -> None: elapsed_ms = int((monotonic() - started) * 1000) tags: dict[str, Any] = {_TAG_ELAPSED_MS: elapsed_ms} - if exc is None: + if exc is None or _is_checkpoint_exception(exc): tags[_TAG_SUCCESS] = True + if exc is not None: + tags[_TAG_CHECKPOINT] = True else: tags[_TAG_SUCCESS] = False tags[_TAG_ERROR_TYPE] = type(exc).__name__ @@ -642,9 +654,12 @@ def finish(self, exc: BaseException | None = None) -> None: try: live_tags: dict[str, Any] - if exc is None: + if exc is None or _is_checkpoint_exception(exc): _mark_success(span) live_tags = {_TAG_SUCCESS: True} + if exc is not None: + span.set_tag(_TAG_CHECKPOINT, value=True) + live_tags[_TAG_CHECKPOINT] = True elif isinstance(exc, HTTPException): _mark_http_exception(span, exc) live_tags = { diff --git a/src/hayhooks/server/utils/deploy_utils.py b/src/hayhooks/server/utils/deploy_utils.py index 4ed4b571..5a2686d9 100644 --- a/src/hayhooks/server/utils/deploy_utils.py +++ b/src/hayhooks/server/utils/deploy_utils.py @@ -33,6 +33,7 @@ from hayhooks.server.pipelines.registry import registry from hayhooks.server.pipelines.sse import SSEStream from hayhooks.server.tracing import ( + SPAN_DURABLE_ATTEMPT, SPAN_PIPELINE_DEPLOY, SPAN_PIPELINE_DEPLOY_COMMIT, SPAN_PIPELINE_DEPLOY_PREPARE, @@ -612,6 +613,41 @@ def _quiesce_idle_durable_deployment(deployment: DurableDeployment, app: FastAPI raise +def _trace_durable_runner( + pipeline_name: str, + revision: str, + kind: str, + runner: Callable, +) -> Callable: + def trace_tags(context: Any) -> dict[str, Any]: + return build_trace_tags( + { + "hayhooks.transport": "durable", + "hayhooks.pipeline.name": pipeline_name, + "hayhooks.durable.execution_id": context.execution_id, + "hayhooks.durable.attempt": context.attempt, + "hayhooks.durable.kind": kind, + "hayhooks.durable.definition_revision": revision, + } + ) + + if inspect.iscoroutinefunction(runner): + + @wraps(runner) + async def traced_async(context: Any, request: BaseModel) -> object: + with trace_operation(SPAN_DURABLE_ATTEMPT, tags=trace_tags(context)): + return await runner(context, request) + + return traced_async + + @wraps(runner) + def traced_sync(context: Any, request: BaseModel) -> object: + with trace_operation(SPAN_DURABLE_ATTEMPT, tags=trace_tags(context)): + return runner(context, request) + + return traced_sync + + def _register_prepared_pipeline( # noqa: C901, PLR0912, PLR0915 pipeline_name: str, pipeline_wrapper: BasePipelineWrapper, @@ -714,7 +750,12 @@ def _register_prepared_pipeline( # noqa: C901, PLR0912, PLR0915 pipeline_wrapper.durable_revision or "", store, request_model, - runner, + _trace_durable_runner( + pipeline_name, + pipeline_wrapper.durable_revision or "", + adapter.kind.value, + runner, + ), kind=adapter.kind, result_model=result_model, resume_model=pipeline_wrapper.durable_resume_model, diff --git a/tests/test_durable_hayhooks.py b/tests/test_durable_hayhooks.py index 10fca859..7bc73493 100644 --- a/tests/test_durable_hayhooks.py +++ b/tests/test_durable_hayhooks.py @@ -22,6 +22,7 @@ from hayhooks.durable.store import ExecutionStoreError, MemoryExecutionStore from hayhooks.server.app import create_app from hayhooks.server.pipelines.registry import registry +from hayhooks.server.tracing import SPAN_DURABLE_ATTEMPT from hayhooks.server.utils import deploy_utils from hayhooks.server.utils.base_pipeline_wrapper import BasePipelineWrapper from hayhooks.server.utils.deploy_utils import commit_prepared_pipeline, deploy_pipeline_files, undeploy_pipeline @@ -111,7 +112,7 @@ async def run_durable_async(self, context: DurableContext, request: Request) -> @pytest.mark.skipif(not _HAYSTACK_V3, reason="Hayhooks durable wrappers require Haystack 3.1+") -def test_hayhooks_mounts_and_runs_a_durable_wrapper(durable_client, wait_for_execution): +def test_hayhooks_mounts_and_runs_a_durable_wrapper(durable_client, wait_for_execution, recording_tracer): commit_prepared_pipeline(PreparedPipeline("durable-job", wrapper_for(DurableWrapper)), app=durable_client.app) headers = {"Idempotency-Key": "retry-after-response-loss"} submitted = durable_client.post("/durable-job/run-durable", json={"value": 7}, headers=headers) @@ -119,6 +120,17 @@ def test_hayhooks_mounts_and_runs_a_durable_wrapper(durable_client, wait_for_exe execution_id = submitted.json()["execution_id"] path = f"/durable-job/executions/{execution_id}" assert wait_for_execution(durable_client, path, "completed")["result"] == {"value": 7} + attempt_span = next(span for span in recording_tracer.spans if span.operation_name == SPAN_DURABLE_ATTEMPT) + assert attempt_span.tags == { + "hayhooks.transport": "durable", + "hayhooks.pipeline.name": "durable-job", + "hayhooks.durable.execution_id": execution_id, + "hayhooks.durable.attempt": 1, + "hayhooks.durable.kind": "pipeline", + "hayhooks.durable.definition_revision": "test-v1", + "hayhooks.success": True, + "hayhooks.elapsed_ms": attempt_span.tags["hayhooks.elapsed_ms"], + } replay = durable_client.post("/durable-job/run-durable", json={"value": 7}, headers=headers) assert replay.status_code == 200 and replay.headers["idempotent-replay"] == "true" assert replay.json()["execution_id"] == execution_id diff --git a/tests/test_tracing.py b/tests/test_tracing.py index a3b59fa9..adaeca68 100644 --- a/tests/test_tracing.py +++ b/tests/test_tracing.py @@ -579,6 +579,26 @@ def test_haystack_component_spans_are_mirrored_when_flag_enabled(recording_trace assert {"key": "haystack.component.name", "value": "prompt_builder"} in child_span["tags"] +def test_haystack_breakpoint_is_a_checkpoint_not_a_dashboard_error(recording_tracer, monkeypatch): + clear_live_traces() + monkeypatch.setattr(settings, "dashboard_trace_include_haystack_spans", True) + assert configure_tracing() is True + checkpoint_exception = type("BreakpointException", (Exception,), {"__module__": "haystack.core.errors"}) + + with ( + trace_operation("hayhooks.durable.attempt", tags={"hayhooks.pipeline.name": "demo"}), + pytest.raises(checkpoint_exception), + haystack_tracer.trace("haystack.component.run", tags={"haystack.component.name": "work"}), + ): + raise checkpoint_exception("Breaking at component work") + + trace = get_recent_traces(since_ms=None, limit=10)[0] + child_tags = trace["root_span"]["children"][0]["tags"] + assert {"key": "hayhooks.checkpoint", "value": "true"} in child_tags + assert {"key": "hayhooks.success", "value": "true"} in child_tags + assert not any(tag["key"].startswith("hayhooks.error.") for tag in trace["tags"]) + + def test_haystack_component_spans_are_mirrored_without_external_tracing(monkeypatch): clear_live_traces() disable_tracing()