From 548001a18d39e47fe11b5f51d95ca91e75468b11 Mon Sep 17 00:00:00 2001 From: Harrison O'Connor-Hoover Date: Thu, 1 Oct 2026 05:47:58 -0400 Subject: [PATCH 1/2] Explain graph changes and recorded run outcomes --- HANDOFF.md | 38 ++- docs/control-contracts.md | 43 ++- docs/decisions.md | 12 + src/dander/control/application.py | 24 +- src/dander/control/bundle.py | 40 ++- src/dander/control/change_preview.py | 266 ++++++++++++++++++ .../contracts/v1/fixtures/capabilities.json | 2 +- .../v1/fixtures/graph-change-preview.json | 1 + .../v1/fixtures/run-explanation.json | 1 + src/dander/control/contracts/v1/manifest.json | 2 +- .../v1/schemas/capabilities.schema.json | 2 +- .../schemas/graph-change-preview.schema.json | 1 + .../v1/schemas/run-explanation.schema.json | 1 + src/dander/control/http.py | 25 +- src/dander/control/models.py | 2 + src/dander/control/run_explanation.py | 194 +++++++++++++ tests/control/test_change_preview.py | 241 ++++++++++++++++ tests/control/test_hosted_control.py | 48 ++++ tests/control/test_oidc_auth.py | 15 +- tests/control/test_run_explanation.py | 203 +++++++++++++ ...-286-guided-change-and-run-explanations.md | 35 +++ 21 files changed, 1165 insertions(+), 31 deletions(-) create mode 100644 src/dander/control/change_preview.py create mode 100644 src/dander/control/contracts/v1/fixtures/graph-change-preview.json create mode 100644 src/dander/control/contracts/v1/fixtures/run-explanation.json create mode 100644 src/dander/control/contracts/v1/schemas/graph-change-preview.schema.json create mode 100644 src/dander/control/contracts/v1/schemas/run-explanation.schema.json create mode 100644 src/dander/control/run_explanation.py create mode 100644 tests/control/test_change_preview.py create mode 100644 tests/control/test_run_explanation.py create mode 100644 tickets/DANDER-286-guided-change-and-run-explanations.md diff --git a/HANDOFF.md b/HANDOFF.md index 88a3e6fd..8817cf1e 100644 --- a/HANDOFF.md +++ b/HANDOFF.md @@ -2,38 +2,36 @@ ## Finished -- Scanned repository structure, duplication, import boundaries, error handling, and test health; recorded scope and remaining priorities in `docs/quality-review-2026-09-30.md`. -- Isolated plugin commands so help and catalog search avoid bootstrap, deployment, and provider modules. -- Shared the existing durable-run identity rule between S3 and PostgreSQL, with storage behavior preserved. -- Updated four dependencies after protected CI identified 20 published advisories; the strict runtime dependency audit is now clean locally. -- Refreshed flagged system packages in runtime/starter, Spark, and OCI Dockerfiles; the runtime OS layer and complete Spark/OCI images pass local scans. +- Added a read-only comparison of an unsaved graph with its exact saved revision, including downstream outputs and write effects. +- Added readable explanations of recorded run outcomes and eligible next actions; unavailable measurements remain unknown. +- Exposed both operations through existing Control authentication and regenerated their public contracts. ## Try It -Run `uv run dander plugins --help` and `uv run dander plugins search incident`. Read the quality review for findings and the limits of the scan. +With Control running and a saved graph selected, POST candidate graph JSON to `/v1/projects/{project}/graphs/{graph}/change-preview` with its saved `If-Match` ETag. GET `/v1/runs/{run_id}/explanation` for a configured lifecycle's run. See `docs/control-contracts.md` for semantics and client compatibility. ## Checks -- Full suite with disposable local PostgreSQL 17: **2,419 passed**, one existing Starlette/httpx deprecation warning. -- Focused CLI/storage/lifecycle suite: **75 passed**; final CLI test-import adjustment: **10 passed**. -- Updated dependencies: **139 authentication/ingestion tests passed**; strict `runtime-all` dependency audit reports **no known vulnerabilities**. -- Runtime OS layer and Spark/OCI images: linux/amd64 builds and Trivy scans passed with **zero fixable high/critical findings**. User/driver/handler checks passed. Scaffold/release/import tests: **15 passed**; OCI image/publication tests: **10 passed**. -- Ruff lint/format, canonical strict typing (**509 files**), Control contract drift, documentation links, and diff checks passed. -- Compared moved command bodies and identity comparisons structurally against the original code; behavior-bearing expressions match. Protected CI and merge results are available on the task's attached quality-review PR. +- 87 focused comparison, explanation, HTTP, OIDC, and contract tests passed. +- Canonical strict typing passed across 513 files. +- Ruff lint/format, generated Control contract drift, and diff checks passed. +- Protected CI and exact-main verification remain pending for this implementation step. No live cloud workload was run. ## Decisions -- Use the existing lightweight command-dispatch pattern; preserve command options, installer pins, and storage schemas. -- Keep provider-specific SQL and recovery explicit; defer broader extraction until changes demonstrate a shared responsibility. +- Keep static graph comparison separate from infrastructure deployment preview; no new persistence or provider reads. +- Explain existing run evidence deterministically, with no new AI service or fabricated diagnosis. +- Deliver real date-scoped output repair and the guided interface as subsequent steps of the same active objective. ## Remaining -- Large CLI/bootstrap and warehouse/deployment modules remain the next maintenance targets; see the review's specific recommendations. -- Curated connector releases remain separately blocked on the requested exception to the single-writable-repository rule. Prepared patches remain in `/tmp/dander-curated-compat-20260930/` and the preserved connector branch. -- Existing test-client deprecation warning remains; no live cloud qualification was rerun. +- Implement bounded output-partition repair from retained raw data, preserving normal extraction progress and unrelated output rows. +- Integrate connect, configure, preview, run, and outcome in Druff; its repository exception is awaiting the user's answer. +- Publish and consume a verified immutable producer artifact for the paired Druff client; existing published clients use their original bundle. +- Complete protected PR, merge, and exact-main CI for each runtime change. ## Review First -- `docs/quality-review-2026-09-30.md` for the assessment and scope. -- `src/dander/cli/plugins_command.py` and `src/dander/cli/entrypoint.py` for command isolation. -- `src/dander/control/orchestration.py` for the shared run identity rule. +- `src/dander/control/change_preview.py` for graph comparison and write explanations. +- `src/dander/control/run_explanation.py` for evidence-based outcome handling. +- `docs/control-contracts.md` for the API and publication boundaries. diff --git a/docs/control-contracts.md b/docs/control-contracts.md index ef044d37..7895d3f2 100644 --- a/docs/control-contracts.md +++ b/docs/control-contracts.md @@ -1,7 +1,7 @@ # Dander Control contract bundle -Status: the complete hosted-resource bundle is published in Dander `0.9.0rc19` and is the -immutable release artifact Druff may consume +Published client baseline: Dander `0.9.0rc19`. The current source bundle also contains unreleased +additions described below; their presence in this checkout does not publish a new client artifact. Dander is the authority for data crossing the future Control API boundary. The deterministic `io.dander.control.contracts/v1` bundle lives in `src/dander/control/contracts/v1` and is included @@ -13,12 +13,13 @@ checkout or treat handwritten browser schemas as authoritative. The bundle contains Draft 2020-12 JSON Schemas and canonical fixtures for graph documents, API errors, connectors, plugins, operations, deployment previews, graph validation, run requests and -status, bounded logs, mutation results, and capabilities. Each schema has a stable URN `$id` and +status, graph change previews, run explanations, bounded logs, mutation results, and capabilities. +Each schema has a stable URN `$id` and uses only self-contained `#/$defs` references. `manifest.json` records every file hash and the bundle SHA-256. The bundle digest excludes the manifest itself, so the same source models always produce the same identity. The current reviewed -digest is: +digest for the published RC19 baseline is: ```text 695791dfda6058d68453d9e146146d5cdda1439d86c40a7ec249cb4e14a12be3 @@ -160,6 +161,40 @@ require false all-provider live parity before hosted authentication work began. These are server-internal storage semantics for DANDER-120. DANDER-121 projects them through the separately named hosted service while preserving `dander graph serve --file` unchanged. +## Read-only change previews and run explanations + +Current source adds two operations for a guided authoring workflow. They use the same Control +authentication, graph limits, and normalized lifecycle as the existing routes: + +| Operation | Request | Result | +| --- | --- | --- | +| `graph.change-preview` | `POST /v1/projects/{project}/graphs/{graph}/change-preview`, candidate graph JSON and the saved graph's `If-Match` ETag | Static differences against that exact saved definition; no save, deployment, or warehouse access | +| `run.explain` | `GET /v1/runs/{run_id}/explanation` | Readable outcome and suggested next action from the run's recorded status | + +Preview **before saving**: the graph store retains the current definition, not revision history. +The result identifies changed nodes/connections, downstream affected outputs, before/after writer +settings, and all outputs a full run will write. Moving a node on the canvas alone does not mark +outputs as data changes. Removing an output stops future writes but does not delete its existing +table; moving its destination retains the previous table. Unknown row counts and costs stay null. +The preview's two content hashes identify what was compared. Saving and running still use existing +revision checks; editing again requires another preview. + +This is a static comparison inside one graph, not a data diff or proof of executability. All +configured outputs execute on a full run; affected outputs are not a selective execution plan. +Existing runtime constraints still apply, including replace-only graph targets. Infrastructure +deployment remains the separate `/deployment-preview` operation and can build/push artifacts. + +Run explanations only show counters when the runtime result summary has been collected. An +execution can succeed before those measurements arrive; missing measurements are not zero rows. +Explanations distinguish skipped work, pending cancellation, failed execution, and confirmed +results without guessing an error cause. Suggested actions honor the current run controls; the +explanation does not perform them. Replay is another execution, not rollback or date-range repair. +Neither operation requires provider-specific browser logic or a new service dependency. + +Editors can request change previews; viewers can read run explanations when a lifecycle is wired. +Regenerate consumers from the immutable producer artifact after its release. The guided Druff +surface and date-scoped repair are separate follow-on work, not delivered by these two routes. + ## Hosted Control service `dander control serve` is distinct from the existing one-file bridge. It exposes configured diff --git a/docs/decisions.md b/docs/decisions.md index 3380a4fc..fa730e90 100644 --- a/docs/decisions.md +++ b/docs/decisions.md @@ -1,5 +1,17 @@ # Engineering Decisions +## 2026-10-01 — Explain pipeline changes and runs through the existing Control boundary + +- Compare an unsaved graph with the exact saved revision in a pure, provider-neutral operation. + Keep infrastructure deployment preview separate; no graph history store, warehouse reads, or + new persistence is needed for an author to review a proposed edit. +- Derive readable run explanations from the existing normalized status and collected result + summary. Missing measurements remain unknown; no generated diagnosis or additional AI service + is introduced. Existing start, cancel, and replay semantics remain unchanged. +- Preserve the complete guided-journey objective: the new routes are foundations for Druff, + while date-scoped repair requires its own verified execution behavior. Do not relabel full + replay as date-range repair or advertise an interface before it is implemented. + ## 2026-09-30 — Keep command loading and run identity at their owning boundaries - Plugin commands live in `cli/plugins_command.py`. The console dispatcher selects their group diff --git a/src/dander/control/application.py b/src/dander/control/application.py index a1840eb0..4f3235be 100644 --- a/src/dander/control/application.py +++ b/src/dander/control/application.py @@ -14,6 +14,7 @@ build_typed_operation_catalog, build_typed_plugin_catalog, ) +from dander.control.change_preview import GraphChangePreviewResponse, compare_graph_changes from dander.control.graph_store import ( MAX_GRAPH_DOCUMENT_BYTES, MAX_GRAPH_PAGE_SIZE, @@ -36,12 +37,14 @@ LogPageResponse, MutationResult, OperationCatalogResponse, + PipelineGraphDocument, PluginCatalogResponse, ProjectListResponse, ProjectSummaryResponse, RunPageResponse, RunStatusResponse, ) +from dander.control.run_explanation import RunExplanationResponse, explain_run if TYPE_CHECKING: from collections.abc import Callable @@ -219,11 +222,14 @@ def capabilities(self) -> CapabilitiesResponse: "graph.edit", "graph.delete", "graph.validate", + "graph.change-preview", ] if self.preview_port is not None: operations.append("deployment.preview") if self.lifecycle_port is not None: - operations.extend(["run.start", "run.read", "run.logs", "run.cancel", "run.replay"]) + operations.extend( + ["run.start", "run.read", "run.explain", "run.logs", "run.cancel", "run.replay"] + ) return CapabilitiesResponse( dander_version=__version__, contract=ContractIdentity(id=BUNDLE_ID, sha256=packaged_bundle_digest()), @@ -299,6 +305,18 @@ def preview_graph( self.require_graph_revision(project, graph, expected_revision) ) + def preview_graph_changes( + self, + project: str, + graph: str, + candidate: PipelineGraphDocument, + *, + expected_revision: str, + ) -> GraphChangePreviewResponse: + """Compare an unsaved candidate with the exact current saved graph.""" + baseline = self.require_graph_revision(project, graph, expected_revision) + return compare_graph_changes(baseline.document, candidate) + def start_run( self, project: str, @@ -342,6 +360,10 @@ def start_run( def get_run(self, address: RunAddress) -> RunStatusResponse: return self._require_lifecycle().get(address) + def explain_run(self, address: RunAddress) -> RunExplanationResponse: + """Explain the same normalized status used by the existing run controls.""" + return explain_run(self.get_run(address)) + def list_runs(self, *, cursor: str | None, limit: int) -> RunPageResponse: if not 1 <= limit <= 100: raise ValueError("The requested run page size is invalid.") diff --git a/src/dander/control/bundle.py b/src/dander/control/bundle.py index 3f5120f9..aa603cf5 100644 --- a/src/dander/control/bundle.py +++ b/src/dander/control/bundle.py @@ -10,6 +10,7 @@ if TYPE_CHECKING: from pydantic import BaseModel +from dander.control.change_preview import GraphChangePreviewResponse, compare_graph_changes from dander.control.models import ( ApiErrorEnvelope, CapabilitiesResponse, @@ -28,8 +29,10 @@ ProjectListResponse, RunPageResponse, RunRequest, + RunState, RunStatusResponse, ) +from dander.control.run_explanation import RunExplanationResponse, explain_run from dander.pipeline.graph import graph_to_payload BUNDLE_ID: Final = "io.dander.control.contracts/v1" @@ -45,6 +48,7 @@ "control-bootstrap": ControlBootstrapDescriptor, "deployment-preview": DeploymentPreviewResponse, "graph-create": GraphCreateRequest, + "graph-change-preview": GraphChangePreviewResponse, "graph-page": GraphPageResponse, "graph-resource": GraphResourceResponse, "graph-validation": GraphValidationResponse, @@ -55,6 +59,7 @@ "plugin-catalog": PluginCatalogResponse, "project-list": ProjectListResponse, "run-page": RunPageResponse, + "run-explanation": RunExplanationResponse, "run-request": RunRequest, "run-status": RunStatusResponse, } @@ -402,7 +407,14 @@ def _fixtures() -> dict[str, tuple[str, dict[str, Any]]]: "minimum_druff_contract": "1.0.0", "maximum_druff_contract": "1.x", }, - "operations": ["graph.read", "graph.edit", "graph.validate", "run.read"], + "operations": [ + "graph.read", + "graph.edit", + "graph.validate", + "graph.change-preview", + "run.read", + "run.explain", + ], "limits": { "max_graph_bytes": 5 * 1024 * 1024, "max_page_size": 100, @@ -623,6 +635,32 @@ def _fixtures() -> dict[str, tuple[str, dict[str, Any]]]: "run-request", {"expected_revision": "opaque-revision", "idempotency_key": "idem-synthetic"}, ), + "graph-change-preview": ( + "graph-change-preview", + compare_graph_changes( + PipelineGraphDocument(name="new_pipeline"), + PipelineGraphDocument.model_validate(_graph_fixture()), + ).model_dump(mode="json"), + ), + "run-explanation": ( + "run-explanation", + explain_run( + RunStatusResponse( + run_id="run-synthetic", + state=RunState.SUCCEEDED, + stage="complete", + result_schema="io.dander.control.execution-result-summary/v1", + endpoints=1, + extracted=2, + affected=2, + models=1, + assertions=1, + assets=1, + can_replay=True, + logs_available=True, + ) + ).model_dump(mode="json"), + ), "run-status": ( "run-status", { diff --git a/src/dander/control/change_preview.py b/src/dander/control/change_preview.py new file mode 100644 index 00000000..5d628375 --- /dev/null +++ b/src/dander/control/change_preview.py @@ -0,0 +1,266 @@ +"""Read-only impact of an unsaved graph compared with its saved definition. + +This is a static authoring preview, not an execution plan or a warehouse data diff. It never +loads provider credentials, reads rows, or saves the candidate. The caller owns revision checks. +""" + +from __future__ import annotations + +import json +from typing import Literal + +from pydantic import Field + +from dander.control.graph_store import canonicalize_graph_document +from dander.control.models import ( + ControlModel, + GraphNodeDocument, + PipelineGraphDocument, + TargetNodeDocument, + WriterDocument, +) +from dander.writer import WriteMode + + +class GraphNodeChange(ControlModel): + node_id: str + name: str + change: Literal["added", "removed", "modified", "layout_only"] + changed_properties: tuple[str, ...] = () + affects_data: bool + + +class GraphConnectionChange(ControlModel): + source: str + target: str + change: Literal["added", "removed", "modified"] + + +class GraphOutputState(ControlModel): + """Only declared output coordinates and field names, never arbitrary node configuration.""" + + writer: WriterDocument | None + field_names: tuple[str, ...] + + +class GraphOutputImpact(ControlModel): + node_id: str + name: str + affected: bool + before: GraphOutputState | None + after: GraphOutputState | None + effect: str + + +class GraphChangeEstimates(ControlModel): + rows_written: None = None + cost_usd: None = None + explanation: str = ( + "Unknown: this preview reads graph definitions only, not source or warehouse data." + ) + + +class GraphChangePreviewResponse(ControlModel): + """Static changes and potential downstream effects; a run still executes all outputs.""" + + baseline_content_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") + candidate_content_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") + graph_properties_changed: tuple[str, ...] + node_changes: tuple[GraphNodeChange, ...] + connection_changes: tuple[GraphConnectionChange, ...] + affected_outputs: tuple[str, ...] + outputs: tuple[GraphOutputImpact, ...] + run_output_ids: tuple[str, ...] + estimates: GraphChangeEstimates = Field(default_factory=GraphChangeEstimates) + limitations: tuple[str, ...] = ( + "Static impact within this graph only; external consumers are not inspected.", + "Affected outputs describe changed definitions, not a selection of outputs to execute. " + "A full run writes every configured output, including unchanged outputs.", + "Declared write behavior is not proof that this graph can execute. Runtime validation " + "and provider checks still apply; the current graph runtime supports replace targets.", + "Saving a graph does not write or delete warehouse data.", + ) + + +def compare_graph_changes( + baseline: PipelineGraphDocument, + candidate: PipelineGraphDocument, +) -> GraphChangePreviewResponse: + """Compare canonical documents without exposing configuration values or touching providers. + + Nodes are matched by stable ID, and connections by endpoint pair. The old and new edges + both participate in downstream impact, so disconnecting or removing a node cannot hide its + previous outputs. Traversal is iterative because the union of two valid DAGs can be cyclic. + """ + before = canonicalize_graph_document(baseline) + after = canonicalize_graph_document(candidate) + old_nodes = {node.id: node for node in before.document.nodes} + new_nodes = {node.id: node for node in after.document.nodes} + node_changes = tuple( + change + for node_id in sorted(old_nodes.keys() | new_nodes.keys()) + if (change := _node_change(node_id, old_nodes.get(node_id), new_nodes.get(node_id))) + is not None + ) + old_edges = _connections(before.document) + new_edges = _connections(after.document) + connection_changes = tuple( + GraphConnectionChange( + source=source, + target=target, + change=( + "added" + if (source, target) not in old_edges + else "removed" + if (source, target) not in new_edges + else "modified" + ), + ) + for source, target in sorted(old_edges.keys() | new_edges.keys()) + if old_edges.get((source, target)) != new_edges.get((source, target)) + ) + graph_properties_changed = tuple( + key + for key in ("name", "trigger") + if getattr(before.document, key) != getattr(after.document, key) + ) + affected = {change.node_id for change in node_changes if change.affects_data} + affected.update(change.target for change in connection_changes) + if "trigger" in graph_properties_changed: + affected.update(old_nodes.keys() | new_nodes.keys()) + successors: dict[str, set[str]] = {} + for source, target in old_edges.keys() | new_edges.keys(): + successors.setdefault(source, set()).add(target) + pending = list(affected) + while pending: + for node_id in successors.get(pending.pop(), set()) - affected: + affected.add(node_id) + pending.append(node_id) + + old_targets = { + node_id: node for node_id, node in old_nodes.items() if isinstance(node, TargetNodeDocument) + } + new_targets = { + node_id: node for node_id, node in new_nodes.items() if isinstance(node, TargetNodeDocument) + } + outputs = tuple( + _output_impact(node_id, old_targets.get(node_id), new_targets.get(node_id), affected) + for node_id in sorted(old_targets.keys() | new_targets.keys()) + ) + return GraphChangePreviewResponse( + baseline_content_sha256=before.content_sha256, + candidate_content_sha256=after.content_sha256, + graph_properties_changed=graph_properties_changed, + node_changes=node_changes, + connection_changes=connection_changes, + affected_outputs=tuple(output.node_id for output in outputs if output.affected), + outputs=outputs, + run_output_ids=tuple(sorted(new_targets)), + ) + + +def _node_change( + node_id: str, + before: GraphNodeDocument | None, + after: GraphNodeDocument | None, +) -> GraphNodeChange | None: + if before is None: + assert after is not None + return GraphNodeChange(node_id=node_id, name=after.name, change="added", affects_data=True) + if after is None: + return GraphNodeChange( + node_id=node_id, name=before.name, change="removed", affects_data=True + ) + old_payload = before.model_dump(mode="json", by_alias=True) + new_payload = after.model_dump(mode="json", by_alias=True) + properties = tuple( + key + for key in sorted(old_payload.keys() | new_payload.keys()) + if old_payload.get(key) != new_payload.get(key) + ) + if not properties: + return None + layout_only = properties == ("visual",) + return GraphNodeChange( + node_id=node_id, + name=after.name, + change="layout_only" if layout_only else "modified", + changed_properties=properties, + affects_data=not layout_only, + ) + + +def _connections(document: PipelineGraphDocument) -> dict[tuple[str, str], tuple[str, ...]]: + grouped: dict[tuple[str, str], list[str]] = {} + for edge in document.edges: + grouped.setdefault((edge.source, edge.target), []).append( + json.dumps(edge.model_dump(mode="json", by_alias=True), sort_keys=True) + ) + return {pair: tuple(sorted(edges)) for pair, edges in grouped.items()} + + +def _output_state(node: TargetNodeDocument | None) -> GraphOutputState | None: + if node is None: + return None + return GraphOutputState( + writer=node.config.writer, field_names=tuple(f.name for f in node.fields) + ) + + +def _output_impact( + node_id: str, + before: TargetNodeDocument | None, + after: TargetNodeDocument | None, + affected: set[str], +) -> GraphOutputImpact: + old_state = _output_state(before) + new_state = _output_state(after) + if after is None: + assert before is not None + effect = ( + "Removed from this graph. Future runs stop writing this output; " + "existing warehouse data is not deleted." + ) + name = before.name + else: + name = after.name + prefix = "New graph output. " if before is None else "" + old_writer = old_state.writer if old_state is not None else None + new_writer = after.config.writer + if old_writer is not None and new_writer is not None: + old_destination = old_writer.destination + new_destination = new_writer.destination + if ( + old_destination.project, + old_destination.dataset, + old_destination.table, + ) != ( + new_destination.project, + new_destination.dataset, + new_destination.table, + ): + prefix = ( + "Destination changed. Existing data at the previous destination is retained. " + ) + effect = prefix + _write_effect(new_writer) + return GraphOutputImpact( + node_id=node_id, + name=name, + affected=node_id in affected, + before=old_state, + after=new_state, + effect=effect, + ) + + +def _write_effect(writer: WriterDocument | None) -> str: + if writer is None: + return "Write behavior is unknown: this target has no declared writer configuration." + effects = { + WriteMode.REPLACE: "Replaces all rows in the destination with this run's result.", + WriteMode.SCD1: "Updates matching business keys and inserts new rows.", + WriteMode.SCD2: "Keeps row history and records new versions of changed business keys.", + WriteMode.SNAPSHOT: "Writes a snapshot using the configured partitioning.", + WriteMode.INCREMENTAL: "Writes incremental rows using the configured cursor and keys.", + } + return "Declared next-run behavior: " + effects[writer.write_mode] diff --git a/src/dander/control/contracts/v1/fixtures/capabilities.json b/src/dander/control/contracts/v1/fixtures/capabilities.json index e13eea21..84265361 100644 --- a/src/dander/control/contracts/v1/fixtures/capabilities.json +++ b/src/dander/control/contracts/v1/fixtures/capabilities.json @@ -1 +1 @@ -{"api_version":"v1","compatibility":{"maximum_druff_contract":"1.x","minimum_druff_contract":"1.0.0"},"contract":{"id":"io.dander.control.contracts/v1","sha256":"0000000000000000000000000000000000000000000000000000000000000000"},"dander_version":"0.0.0","limits":{"max_graph_bytes":5242880,"max_log_records":500,"max_page_size":100},"operations":["graph.read","graph.edit","graph.validate","run.read"]} +{"api_version":"v1","compatibility":{"maximum_druff_contract":"1.x","minimum_druff_contract":"1.0.0"},"contract":{"id":"io.dander.control.contracts/v1","sha256":"0000000000000000000000000000000000000000000000000000000000000000"},"dander_version":"0.0.0","limits":{"max_graph_bytes":5242880,"max_log_records":500,"max_page_size":100},"operations":["graph.read","graph.edit","graph.validate","graph.change-preview","run.read","run.explain"]} diff --git a/src/dander/control/contracts/v1/fixtures/graph-change-preview.json b/src/dander/control/contracts/v1/fixtures/graph-change-preview.json new file mode 100644 index 00000000..84beba89 --- /dev/null +++ b/src/dander/control/contracts/v1/fixtures/graph-change-preview.json @@ -0,0 +1 @@ +{"affected_outputs":["target_incremental","target_replace","target_scd1","target_scd2","target_snapshot"],"baseline_content_sha256":"97cb66a56e300fbcba2ee316511bd1bc14ffe115903f6a4d61452e236e9a19a1","candidate_content_sha256":"8d4b78b016880a28a55d2f7f8949616e617bc141abc4637afd2bb4015f634f2a","connection_changes":[{"change":"added","source":"normalize_records","target":"target_incremental"},{"change":"added","source":"normalize_records","target":"target_replace"},{"change":"added","source":"normalize_records","target":"target_scd1"},{"change":"added","source":"normalize_records","target":"target_scd2"},{"change":"added","source":"normalize_records","target":"target_snapshot"},{"change":"added","source":"source_records","target":"normalize_records"},{"change":"added","source":"source_regions","target":"normalize_records"}],"estimates":{"cost_usd":null,"explanation":"Unknown: this preview reads graph definitions only, not source or warehouse data.","rows_written":null},"graph_properties_changed":["name","trigger"],"limitations":["Static impact within this graph only; external consumers are not inspected.","Affected outputs describe changed definitions, not a selection of outputs to execute. A full run writes every configured output, including unchanged outputs.","Declared write behavior is not proof that this graph can execute. Runtime validation and provider checks still apply; the current graph runtime supports replace targets.","Saving a graph does not write or delete warehouse data."],"node_changes":[{"affects_data":true,"change":"added","changed_properties":[],"name":"Normalize records","node_id":"normalize_records"},{"affects_data":true,"change":"added","changed_properties":[],"name":"Notify operator","node_id":"notify_operator"},{"affects_data":true,"change":"added","changed_properties":[],"name":"Records","node_id":"source_records"},{"affects_data":true,"change":"added","changed_properties":[],"name":"Regions","node_id":"source_regions"},{"affects_data":true,"change":"added","changed_properties":[],"name":"Target Incremental","node_id":"target_incremental"},{"affects_data":true,"change":"added","changed_properties":[],"name":"Target Replace","node_id":"target_replace"},{"affects_data":true,"change":"added","changed_properties":[],"name":"Target Scd1","node_id":"target_scd1"},{"affects_data":true,"change":"added","changed_properties":[],"name":"Target Scd2","node_id":"target_scd2"},{"affects_data":true,"change":"added","changed_properties":[],"name":"Target Snapshot","node_id":"target_snapshot"}],"outputs":[{"affected":true,"after":{"field_names":["id","updated_at"],"writer":{"clustering":[],"cursor_field":"updated_at","destination":{"business_key":["id"],"dataset":"analytics","project":null,"table":"target_incremental"},"max_batch_rows":10000,"partitioning":null,"schema_evolution":"strict","transport":"storage_write","write_mode":"incremental"}},"before":null,"effect":"New graph output. Declared next-run behavior: Writes incremental rows using the configured cursor and keys.","name":"Target Incremental","node_id":"target_incremental"},{"affected":true,"after":{"field_names":["id","updated_at"],"writer":{"clustering":[],"cursor_field":null,"destination":{"business_key":[],"dataset":"analytics","project":null,"table":"target_replace"},"max_batch_rows":10000,"partitioning":null,"schema_evolution":"strict","transport":"copy","write_mode":"replace"}},"before":null,"effect":"New graph output. Declared next-run behavior: Replaces all rows in the destination with this run's result.","name":"Target Replace","node_id":"target_replace"},{"affected":true,"after":{"field_names":["id","updated_at"],"writer":{"clustering":[],"cursor_field":null,"destination":{"business_key":["id"],"dataset":"analytics","project":null,"table":"target_scd1"},"max_batch_rows":10000,"partitioning":null,"schema_evolution":"strict","transport":"load_job","write_mode":"scd1"}},"before":null,"effect":"New graph output. Declared next-run behavior: Updates matching business keys and inserts new rows.","name":"Target Scd1","node_id":"target_scd1"},{"affected":true,"after":{"field_names":["id","updated_at"],"writer":{"clustering":[],"cursor_field":null,"destination":{"business_key":["id"],"dataset":"analytics","project":null,"table":"target_scd2"},"max_batch_rows":10000,"partitioning":null,"schema_evolution":"strict","transport":"copy","write_mode":"scd2"}},"before":null,"effect":"New graph output. Declared next-run behavior: Keeps row history and records new versions of changed business keys.","name":"Target Scd2","node_id":"target_scd2"},{"affected":true,"after":{"field_names":["id","updated_at"],"writer":{"clustering":[],"cursor_field":null,"destination":{"business_key":[],"dataset":"analytics","project":null,"table":"target_snapshot"},"max_batch_rows":10000,"partitioning":null,"schema_evolution":"strict","transport":"load_job","write_mode":"snapshot"}},"before":null,"effect":"New graph output. Declared next-run behavior: Writes a snapshot using the configured partitioning.","name":"Target Snapshot","node_id":"target_snapshot"}],"run_output_ids":["target_incremental","target_replace","target_scd1","target_scd2","target_snapshot"]} diff --git a/src/dander/control/contracts/v1/fixtures/run-explanation.json b/src/dander/control/contracts/v1/fixtures/run-explanation.json new file mode 100644 index 00000000..92021aaf --- /dev/null +++ b/src/dander/control/contracts/v1/fixtures/run-explanation.json @@ -0,0 +1 @@ +{"available_actions":["refresh","logs","replay","inspect_outputs"],"caveats":["Recorded counts describe this run; they are not a before/after data comparison.","Runtime telemetry has not been collected for this run.","Replay starts another execution; it does not undo prior writes or repair only a selected date range."],"details":["Recorded extraction: 2 rows from 1 endpoints.","Recorded output: 2 affected rows and 1 models.","Recorded validation and catalog: 1 assertions evaluated and 1 assets."],"next_action":{"kind":"inspect_outputs","label":"Review the pipeline outputs","reason":"Confirm that the recorded output matches the result you expected."},"results_available":true,"run_id":"run-synthetic","state":"succeeded","summary":"The run succeeded and its result measurements are available."} diff --git a/src/dander/control/contracts/v1/manifest.json b/src/dander/control/contracts/v1/manifest.json index a90c0661..3ea99965 100644 --- a/src/dander/control/contracts/v1/manifest.json +++ b/src/dander/control/contracts/v1/manifest.json @@ -1 +1 @@ -{"bundle_id":"io.dander.control.contracts/v1","bundle_sha256":"bddaca593832c1145409dfb1745cf67d0bd0d62531061113d06055b71f5fa5b1","files":[{"contract":"api-error","kind":"fixture","path":"fixtures/api-error.json","sha256":"ed11f73d577b895d9cb6c1167e696b95a30b8ffa2c2d784ff83b7c6346b6ca0d"},{"contract":"capabilities","kind":"fixture","path":"fixtures/capabilities.json","sha256":"7adeda8a3addfdc492d5cb196a71b5f75abaaf675e504584ae40e07e48b8f438"},{"contract":"connector-catalog","kind":"fixture","path":"fixtures/connector-catalog.json","sha256":"c5d8c023bebdc7f39d445bbf8e54f8b3c2f0a72f57aaff3e5c8334274843dc25"},{"contract":"control-bootstrap","kind":"fixture","path":"fixtures/control-bootstrap.json","sha256":"4dac4acea0153c63bf6fbefc6a2f25a4354c1f61f448af3330adf2d1a95d9c80"},{"contract":"deployment-preview","kind":"fixture","path":"fixtures/deployment-preview.json","sha256":"aa9703ab7e8e1ef91a1f8324798709b01bbc5580ba14e3b936fe882820fba539"},{"contract":"graph-create","kind":"fixture","path":"fixtures/graph-create.json","sha256":"c576c1e3109e9dbbdb3f9c81aec477e84639458dbfc2846e1f48f8f3331cc529"},{"contract":"graph-page","kind":"fixture","path":"fixtures/graph-page.json","sha256":"95c2133bb9b2c5aa18de13b73e91322450fcdbf958bee619b603276ac88ba816"},{"contract":"graph-resource","kind":"fixture","path":"fixtures/graph-resource.json","sha256":"68b523de7a34f8383f6d32e47587669091e1292dc93d5afe6b978c7b01dc811d"},{"contract":"graph-validation","kind":"fixture","path":"fixtures/graph-validation.json","sha256":"64a5f8a1f39bde3ef30f5401cf285a9de6091aad9771591b1817168e646dc3e0"},{"contract":"log-page","kind":"fixture","path":"fixtures/log-page.json","sha256":"481c46c6365cfdcaf184dacd53f59cb34e894ec9a11675e876bd220aa609d76c"},{"contract":"mutation-result","kind":"fixture","path":"fixtures/mutation-result.json","sha256":"aa0eb0a3e48f03102c1165aff9933e2be1c54f80a2e29bb67a0aa3bc1fae9898"},{"contract":"operation-catalog","kind":"fixture","path":"fixtures/operation-catalog.json","sha256":"adb73dac2061e1b00a85148cfaf810040aeafb32fb7fd670c077596853f05508"},{"contract":"pipeline-graph","kind":"fixture","path":"fixtures/pipeline-graph-alias-input.json","sha256":"de73bc8e172dfa8daba536fe14fef13a272dd61e8f1bcfef286fe043477ca451"},{"contract":"pipeline-graph","kind":"fixture","path":"fixtures/pipeline-graph.json","sha256":"65c6fa7b88221b1e9a1e51ec773fb9c374533d0ed54e2f2a5c8c7a1242b1b1bc"},{"contract":"plugin-catalog","kind":"fixture","path":"fixtures/plugin-catalog.json","sha256":"039cf19f5ce4c4d5b35e8b1801ff151cf0912b7f90792196dce76e2fce7726df"},{"contract":"project-list","kind":"fixture","path":"fixtures/project-list.json","sha256":"157e70424be88bfaf0429d66dd3a8103a4e402e3f9067685bfe78e2faadf56d3"},{"contract":"run-page","kind":"fixture","path":"fixtures/run-page.json","sha256":"cae943a95060651795160b980d8c25b99ff92440cf905b115b96282cf624fd87"},{"contract":"run-request","kind":"fixture","path":"fixtures/run-request.json","sha256":"36013392b347194d13be6747840c93edb9c8dbf19468a6e19032f26899824f71"},{"contract":"run-status","kind":"fixture","path":"fixtures/run-status.json","sha256":"c2d76c2b9879b45d954027247da0849f216db44d8f449ea4a0c24443a8283556"},{"contract":"api-error","kind":"schema","path":"schemas/api-error.schema.json","sha256":"de86daa1616237146b0666f3c69bf7c22d5679595d756ca980becb69b1613164"},{"contract":"capabilities","kind":"schema","path":"schemas/capabilities.schema.json","sha256":"a9d85b8a6e5f73a425432f5f436e7809592889e72c71762c5c544151a1bf1217"},{"contract":"connector-catalog","kind":"schema","path":"schemas/connector-catalog.schema.json","sha256":"132f169721b81569c4934d03c34e1a276176b40a45711de61264bb4928420dc5"},{"contract":"control-bootstrap","kind":"schema","path":"schemas/control-bootstrap.schema.json","sha256":"013a30d4d4c6f417a898ce0c81fc00dcbfc138e76b79d49694d1af7a6f8b4096"},{"contract":"deployment-preview","kind":"schema","path":"schemas/deployment-preview.schema.json","sha256":"67449ad10f78b972ff3fbda98948f3679dcdecf908ae468fe8ddf843158dfe81"},{"contract":"graph-create","kind":"schema","path":"schemas/graph-create.schema.json","sha256":"15cefbe92bdc171916d0118bfa33c665ac39a409661b222861a6f72a23331b41"},{"contract":"graph-page","kind":"schema","path":"schemas/graph-page.schema.json","sha256":"204018767f05e10a110f7caa177312ffea1266cfbd7eb841ed8a8ff90fbe1e09"},{"contract":"graph-resource","kind":"schema","path":"schemas/graph-resource.schema.json","sha256":"dc3ad0085768407720c825c95a5ff7f1ea01e85f6377a8f32445e926a9e88c61"},{"contract":"graph-validation","kind":"schema","path":"schemas/graph-validation.schema.json","sha256":"e835cd4ed7f698420a643305116f46644d54c8514548b849e34c23d398123d54"},{"contract":"log-page","kind":"schema","path":"schemas/log-page.schema.json","sha256":"bdc4059aecaeb7e18c93b3173726b256119306137a0988ddf1673bb50d260bc6"},{"contract":"mutation-result","kind":"schema","path":"schemas/mutation-result.schema.json","sha256":"078ca417acce9d4cd61c89607e5174156e986f432055c07509b9282fe2f537b7"},{"contract":"operation-catalog","kind":"schema","path":"schemas/operation-catalog.schema.json","sha256":"3ed2d3fd7defc677f7841b4e252a31dffd739922829d572ba9d5a7a635fae000"},{"contract":"pipeline-graph","kind":"schema","path":"schemas/pipeline-graph.schema.json","sha256":"3c2aa99c1ab5a93f2c590c0634fd1b39b8d1e84c56fbe8e42e0ff3051ee40377"},{"contract":"plugin-catalog","kind":"schema","path":"schemas/plugin-catalog.schema.json","sha256":"d934e93c333123284d96e2c9fb19080c952c6e353a721207d36228347923eb83"},{"contract":"project-list","kind":"schema","path":"schemas/project-list.schema.json","sha256":"a61e8c0dff565792079c1954584d7424a90fad35de85745682492a60e8067243"},{"contract":"run-page","kind":"schema","path":"schemas/run-page.schema.json","sha256":"6eb580e2803181117c7e52b61d46d183f291865c8d147369cba0488de06797dc"},{"contract":"run-request","kind":"schema","path":"schemas/run-request.schema.json","sha256":"0a8aa50cf4e82b573476e446e1c9eb6fba1482625abb300b719199ce68866eba"},{"contract":"run-status","kind":"schema","path":"schemas/run-status.schema.json","sha256":"d29fbb9d1614a5ea27a762dab58089c7b352a4cd2f8d27e019e8896237359b07"}],"json_schema_dialect":"https://json-schema.org/draft/2020-12/schema","reference_convention":"self-contained internal #/$defs references","schema":"io.dander.control.contracts-manifest/v1"} +{"bundle_id":"io.dander.control.contracts/v1","bundle_sha256":"03fdac1a19a98f10f1af6fa7821b705291833f88a23e723300abace660130d5b","files":[{"contract":"api-error","kind":"fixture","path":"fixtures/api-error.json","sha256":"ed11f73d577b895d9cb6c1167e696b95a30b8ffa2c2d784ff83b7c6346b6ca0d"},{"contract":"capabilities","kind":"fixture","path":"fixtures/capabilities.json","sha256":"91dff64c697c6a527c9e71f2b67e167eea434e19b28a884bce83e390ce94c7dd"},{"contract":"connector-catalog","kind":"fixture","path":"fixtures/connector-catalog.json","sha256":"c5d8c023bebdc7f39d445bbf8e54f8b3c2f0a72f57aaff3e5c8334274843dc25"},{"contract":"control-bootstrap","kind":"fixture","path":"fixtures/control-bootstrap.json","sha256":"4dac4acea0153c63bf6fbefc6a2f25a4354c1f61f448af3330adf2d1a95d9c80"},{"contract":"deployment-preview","kind":"fixture","path":"fixtures/deployment-preview.json","sha256":"aa9703ab7e8e1ef91a1f8324798709b01bbc5580ba14e3b936fe882820fba539"},{"contract":"graph-change-preview","kind":"fixture","path":"fixtures/graph-change-preview.json","sha256":"4c108460b9567164472b9f1cad035b23b0a07da31f2062a0de5cfe1f91b79333"},{"contract":"graph-create","kind":"fixture","path":"fixtures/graph-create.json","sha256":"c576c1e3109e9dbbdb3f9c81aec477e84639458dbfc2846e1f48f8f3331cc529"},{"contract":"graph-page","kind":"fixture","path":"fixtures/graph-page.json","sha256":"95c2133bb9b2c5aa18de13b73e91322450fcdbf958bee619b603276ac88ba816"},{"contract":"graph-resource","kind":"fixture","path":"fixtures/graph-resource.json","sha256":"68b523de7a34f8383f6d32e47587669091e1292dc93d5afe6b978c7b01dc811d"},{"contract":"graph-validation","kind":"fixture","path":"fixtures/graph-validation.json","sha256":"64a5f8a1f39bde3ef30f5401cf285a9de6091aad9771591b1817168e646dc3e0"},{"contract":"log-page","kind":"fixture","path":"fixtures/log-page.json","sha256":"481c46c6365cfdcaf184dacd53f59cb34e894ec9a11675e876bd220aa609d76c"},{"contract":"mutation-result","kind":"fixture","path":"fixtures/mutation-result.json","sha256":"aa0eb0a3e48f03102c1165aff9933e2be1c54f80a2e29bb67a0aa3bc1fae9898"},{"contract":"operation-catalog","kind":"fixture","path":"fixtures/operation-catalog.json","sha256":"adb73dac2061e1b00a85148cfaf810040aeafb32fb7fd670c077596853f05508"},{"contract":"pipeline-graph","kind":"fixture","path":"fixtures/pipeline-graph-alias-input.json","sha256":"de73bc8e172dfa8daba536fe14fef13a272dd61e8f1bcfef286fe043477ca451"},{"contract":"pipeline-graph","kind":"fixture","path":"fixtures/pipeline-graph.json","sha256":"65c6fa7b88221b1e9a1e51ec773fb9c374533d0ed54e2f2a5c8c7a1242b1b1bc"},{"contract":"plugin-catalog","kind":"fixture","path":"fixtures/plugin-catalog.json","sha256":"039cf19f5ce4c4d5b35e8b1801ff151cf0912b7f90792196dce76e2fce7726df"},{"contract":"project-list","kind":"fixture","path":"fixtures/project-list.json","sha256":"157e70424be88bfaf0429d66dd3a8103a4e402e3f9067685bfe78e2faadf56d3"},{"contract":"run-explanation","kind":"fixture","path":"fixtures/run-explanation.json","sha256":"e0ccfd75142882f91fcc0e352b09c5ecf92193646c3a98107f56fa6482b15cd4"},{"contract":"run-page","kind":"fixture","path":"fixtures/run-page.json","sha256":"cae943a95060651795160b980d8c25b99ff92440cf905b115b96282cf624fd87"},{"contract":"run-request","kind":"fixture","path":"fixtures/run-request.json","sha256":"36013392b347194d13be6747840c93edb9c8dbf19468a6e19032f26899824f71"},{"contract":"run-status","kind":"fixture","path":"fixtures/run-status.json","sha256":"c2d76c2b9879b45d954027247da0849f216db44d8f449ea4a0c24443a8283556"},{"contract":"api-error","kind":"schema","path":"schemas/api-error.schema.json","sha256":"de86daa1616237146b0666f3c69bf7c22d5679595d756ca980becb69b1613164"},{"contract":"capabilities","kind":"schema","path":"schemas/capabilities.schema.json","sha256":"3be0992ebd1cbbb0d4dfb2c1205b715a7e4c6131b8a4f2ebf47564b4c354ae0c"},{"contract":"connector-catalog","kind":"schema","path":"schemas/connector-catalog.schema.json","sha256":"132f169721b81569c4934d03c34e1a276176b40a45711de61264bb4928420dc5"},{"contract":"control-bootstrap","kind":"schema","path":"schemas/control-bootstrap.schema.json","sha256":"013a30d4d4c6f417a898ce0c81fc00dcbfc138e76b79d49694d1af7a6f8b4096"},{"contract":"deployment-preview","kind":"schema","path":"schemas/deployment-preview.schema.json","sha256":"67449ad10f78b972ff3fbda98948f3679dcdecf908ae468fe8ddf843158dfe81"},{"contract":"graph-change-preview","kind":"schema","path":"schemas/graph-change-preview.schema.json","sha256":"e45220e046d17c5b608da11e59bbeec6160f56f910b568bd15c8651465e585bf"},{"contract":"graph-create","kind":"schema","path":"schemas/graph-create.schema.json","sha256":"15cefbe92bdc171916d0118bfa33c665ac39a409661b222861a6f72a23331b41"},{"contract":"graph-page","kind":"schema","path":"schemas/graph-page.schema.json","sha256":"204018767f05e10a110f7caa177312ffea1266cfbd7eb841ed8a8ff90fbe1e09"},{"contract":"graph-resource","kind":"schema","path":"schemas/graph-resource.schema.json","sha256":"dc3ad0085768407720c825c95a5ff7f1ea01e85f6377a8f32445e926a9e88c61"},{"contract":"graph-validation","kind":"schema","path":"schemas/graph-validation.schema.json","sha256":"e835cd4ed7f698420a643305116f46644d54c8514548b849e34c23d398123d54"},{"contract":"log-page","kind":"schema","path":"schemas/log-page.schema.json","sha256":"bdc4059aecaeb7e18c93b3173726b256119306137a0988ddf1673bb50d260bc6"},{"contract":"mutation-result","kind":"schema","path":"schemas/mutation-result.schema.json","sha256":"078ca417acce9d4cd61c89607e5174156e986f432055c07509b9282fe2f537b7"},{"contract":"operation-catalog","kind":"schema","path":"schemas/operation-catalog.schema.json","sha256":"3ed2d3fd7defc677f7841b4e252a31dffd739922829d572ba9d5a7a635fae000"},{"contract":"pipeline-graph","kind":"schema","path":"schemas/pipeline-graph.schema.json","sha256":"3c2aa99c1ab5a93f2c590c0634fd1b39b8d1e84c56fbe8e42e0ff3051ee40377"},{"contract":"plugin-catalog","kind":"schema","path":"schemas/plugin-catalog.schema.json","sha256":"d934e93c333123284d96e2c9fb19080c952c6e353a721207d36228347923eb83"},{"contract":"project-list","kind":"schema","path":"schemas/project-list.schema.json","sha256":"a61e8c0dff565792079c1954584d7424a90fad35de85745682492a60e8067243"},{"contract":"run-explanation","kind":"schema","path":"schemas/run-explanation.schema.json","sha256":"e00ca3f82bfca14e77d61f00b14bc2c314820b680ff357b3c45e988eae6a8324"},{"contract":"run-page","kind":"schema","path":"schemas/run-page.schema.json","sha256":"6eb580e2803181117c7e52b61d46d183f291865c8d147369cba0488de06797dc"},{"contract":"run-request","kind":"schema","path":"schemas/run-request.schema.json","sha256":"0a8aa50cf4e82b573476e446e1c9eb6fba1482625abb300b719199ce68866eba"},{"contract":"run-status","kind":"schema","path":"schemas/run-status.schema.json","sha256":"d29fbb9d1614a5ea27a762dab58089c7b352a4cd2f8d27e019e8896237359b07"}],"json_schema_dialect":"https://json-schema.org/draft/2020-12/schema","reference_convention":"self-contained internal #/$defs references","schema":"io.dander.control.contracts-manifest/v1"} diff --git a/src/dander/control/contracts/v1/schemas/capabilities.schema.json b/src/dander/control/contracts/v1/schemas/capabilities.schema.json index 5b00d507..0555601f 100644 --- a/src/dander/control/contracts/v1/schemas/capabilities.schema.json +++ b/src/dander/control/contracts/v1/schemas/capabilities.schema.json @@ -1 +1 @@ -{"$defs":{"CompatibilityRange":{"additionalProperties":false,"properties":{"maximum_druff_contract":{"title":"Maximum Druff Contract","type":"string"},"minimum_druff_contract":{"title":"Minimum Druff Contract","type":"string"}},"required":["minimum_druff_contract","maximum_druff_contract"],"title":"CompatibilityRange","type":"object"},"ContractIdentity":{"additionalProperties":false,"properties":{"id":{"const":"io.dander.control.contracts/v1","title":"Id","type":"string"},"sha256":{"pattern":"^[0-9a-f]{64}$","title":"Sha256","type":"string"}},"required":["id","sha256"],"title":"ContractIdentity","type":"object"},"ControlLimits":{"additionalProperties":false,"properties":{"max_graph_bytes":{"exclusiveMinimum":0,"title":"Max Graph Bytes","type":"integer"},"max_log_records":{"exclusiveMinimum":0,"title":"Max Log Records","type":"integer"},"max_page_size":{"exclusiveMinimum":0,"title":"Max Page Size","type":"integer"}},"required":["max_graph_bytes","max_page_size","max_log_records"],"title":"ControlLimits","type":"object"}},"$id":"urn:dander:control:contracts:v1:capabilities","$schema":"https://json-schema.org/draft/2020-12/schema","additionalProperties":false,"properties":{"api_version":{"const":"v1","default":"v1","title":"Api Version","type":"string"},"compatibility":{"$ref":"#/$defs/CompatibilityRange"},"contract":{"$ref":"#/$defs/ContractIdentity"},"dander_version":{"title":"Dander Version","type":"string"},"limits":{"$ref":"#/$defs/ControlLimits"},"operations":{"items":{"enum":["graph.read","graph.edit","graph.delete","graph.validate","deployment.preview","run.start","run.read","run.logs","run.cancel","run.replay"],"type":"string"},"title":"Operations","type":"array"}},"required":["dander_version","contract","compatibility","operations","limits"],"title":"CapabilitiesResponse","type":"object"} +{"$defs":{"CompatibilityRange":{"additionalProperties":false,"properties":{"maximum_druff_contract":{"title":"Maximum Druff Contract","type":"string"},"minimum_druff_contract":{"title":"Minimum Druff Contract","type":"string"}},"required":["minimum_druff_contract","maximum_druff_contract"],"title":"CompatibilityRange","type":"object"},"ContractIdentity":{"additionalProperties":false,"properties":{"id":{"const":"io.dander.control.contracts/v1","title":"Id","type":"string"},"sha256":{"pattern":"^[0-9a-f]{64}$","title":"Sha256","type":"string"}},"required":["id","sha256"],"title":"ContractIdentity","type":"object"},"ControlLimits":{"additionalProperties":false,"properties":{"max_graph_bytes":{"exclusiveMinimum":0,"title":"Max Graph Bytes","type":"integer"},"max_log_records":{"exclusiveMinimum":0,"title":"Max Log Records","type":"integer"},"max_page_size":{"exclusiveMinimum":0,"title":"Max Page Size","type":"integer"}},"required":["max_graph_bytes","max_page_size","max_log_records"],"title":"ControlLimits","type":"object"}},"$id":"urn:dander:control:contracts:v1:capabilities","$schema":"https://json-schema.org/draft/2020-12/schema","additionalProperties":false,"properties":{"api_version":{"const":"v1","default":"v1","title":"Api Version","type":"string"},"compatibility":{"$ref":"#/$defs/CompatibilityRange"},"contract":{"$ref":"#/$defs/ContractIdentity"},"dander_version":{"title":"Dander Version","type":"string"},"limits":{"$ref":"#/$defs/ControlLimits"},"operations":{"items":{"enum":["graph.read","graph.edit","graph.delete","graph.validate","graph.change-preview","deployment.preview","run.start","run.read","run.explain","run.logs","run.cancel","run.replay"],"type":"string"},"title":"Operations","type":"array"}},"required":["dander_version","contract","compatibility","operations","limits"],"title":"CapabilitiesResponse","type":"object"} diff --git a/src/dander/control/contracts/v1/schemas/graph-change-preview.schema.json b/src/dander/control/contracts/v1/schemas/graph-change-preview.schema.json new file mode 100644 index 00000000..d77ffa9a --- /dev/null +++ b/src/dander/control/contracts/v1/schemas/graph-change-preview.schema.json @@ -0,0 +1 @@ +{"$defs":{"DestinationDocument":{"additionalProperties":false,"properties":{"business_key":{"items":{"type":"string"},"title":"Business Key","type":"array"},"dataset":{"minLength":1,"title":"Dataset","type":"string"},"project":{"anyOf":[{"type":"string"},{"type":"null"}],"default":null,"title":"Project"},"table":{"minLength":1,"title":"Table","type":"string"}},"required":["dataset","table"],"title":"DestinationDocument","type":"object"},"GraphChangeEstimates":{"additionalProperties":false,"properties":{"cost_usd":{"default":null,"title":"Cost Usd","type":"null"},"explanation":{"default":"Unknown: this preview reads graph definitions only, not source or warehouse data.","title":"Explanation","type":"string"},"rows_written":{"default":null,"title":"Rows Written","type":"null"}},"title":"GraphChangeEstimates","type":"object"},"GraphConnectionChange":{"additionalProperties":false,"properties":{"change":{"enum":["added","removed","modified"],"title":"Change","type":"string"},"source":{"title":"Source","type":"string"},"target":{"title":"Target","type":"string"}},"required":["source","target","change"],"title":"GraphConnectionChange","type":"object"},"GraphNodeChange":{"additionalProperties":false,"properties":{"affects_data":{"title":"Affects Data","type":"boolean"},"change":{"enum":["added","removed","modified","layout_only"],"title":"Change","type":"string"},"changed_properties":{"default":[],"items":{"type":"string"},"title":"Changed Properties","type":"array"},"name":{"title":"Name","type":"string"},"node_id":{"title":"Node Id","type":"string"}},"required":["node_id","name","change","affects_data"],"title":"GraphNodeChange","type":"object"},"GraphOutputImpact":{"additionalProperties":false,"properties":{"affected":{"title":"Affected","type":"boolean"},"after":{"anyOf":[{"$ref":"#/$defs/GraphOutputState"},{"type":"null"}]},"before":{"anyOf":[{"$ref":"#/$defs/GraphOutputState"},{"type":"null"}]},"effect":{"title":"Effect","type":"string"},"name":{"title":"Name","type":"string"},"node_id":{"title":"Node Id","type":"string"}},"required":["node_id","name","affected","before","after","effect"],"title":"GraphOutputImpact","type":"object"},"GraphOutputState":{"additionalProperties":false,"description":"Only declared output coordinates and field names, never arbitrary node configuration.","properties":{"field_names":{"items":{"type":"string"},"title":"Field Names","type":"array"},"writer":{"anyOf":[{"$ref":"#/$defs/WriterDocument"},{"type":"null"}]}},"required":["writer","field_names"],"title":"GraphOutputState","type":"object"},"PartitioningDocument":{"additionalProperties":false,"properties":{"field":{"anyOf":[{"type":"string"},{"type":"null"}],"default":null,"title":"Field"},"granularity":{"$ref":"#/$defs/PartitioningType","default":"day"},"require_partition_filter":{"default":false,"title":"Require Partition Filter","type":"boolean"}},"title":"PartitioningDocument","type":"object"},"PartitioningType":{"description":"The closed set of time-unit partitioning granularities a `PartitioningSpec` may declare.\n\nA `StrEnum` (matching the `WriteMode`/`TransformationKind`/`JoinType` convention elsewhere in\n`dander.pipeline`), so it serializes to/from its plain string value stably in YAML and JSON;\nan out-of-set value fails validation with a clear `ValidationError`. Scope is deliberately\nlimited to BigQuery time-unit partitioning — integer-range partitioning is a deferred future\nmember (see `steering/02-engineering.md` on avoiding speculative generality).\n\nAttributes:\n HOUR: Hourly partitions.\n DAY: Daily partitions — the common case and BigQuery's default granularity.\n MONTH: Monthly partitions.\n YEAR: Yearly partitions.","enum":["hour","day","month","year"],"title":"PartitioningType","type":"string"},"SchemaEvolution":{"description":"How a writer handles declared columns absent from an existing target.","enum":["strict","additive"],"title":"SchemaEvolution","type":"string"},"WriteMode":{"description":"Supported load strategies.","enum":["scd1","scd2","snapshot","incremental","replace"],"title":"WriteMode","type":"string"},"WriterDocument":{"additionalProperties":false,"properties":{"clustering":{"items":{"type":"string"},"maxItems":4,"title":"Clustering","type":"array"},"cursor_field":{"anyOf":[{"type":"string"},{"type":"null"}],"default":null,"title":"Cursor Field"},"destination":{"$ref":"#/$defs/DestinationDocument"},"max_batch_rows":{"default":10000,"exclusiveMinimum":0,"maximum":100000,"title":"Max Batch Rows","type":"integer"},"partitioning":{"anyOf":[{"$ref":"#/$defs/PartitioningDocument"},{"type":"null"}],"default":null},"schema_evolution":{"$ref":"#/$defs/SchemaEvolution","default":"strict"},"transport":{"default":"load_job","enum":["load_job","storage_write","copy"],"title":"Transport","type":"string"},"write_mode":{"$ref":"#/$defs/WriteMode"}},"required":["write_mode","destination"],"title":"WriterDocument","type":"object"}},"$id":"urn:dander:control:contracts:v1:graph-change-preview","$schema":"https://json-schema.org/draft/2020-12/schema","additionalProperties":false,"description":"Static changes and potential downstream effects; a run still executes all outputs.","properties":{"affected_outputs":{"items":{"type":"string"},"title":"Affected Outputs","type":"array"},"baseline_content_sha256":{"pattern":"^[0-9a-f]{64}$","title":"Baseline Content Sha256","type":"string"},"candidate_content_sha256":{"pattern":"^[0-9a-f]{64}$","title":"Candidate Content Sha256","type":"string"},"connection_changes":{"items":{"$ref":"#/$defs/GraphConnectionChange"},"title":"Connection Changes","type":"array"},"estimates":{"$ref":"#/$defs/GraphChangeEstimates"},"graph_properties_changed":{"items":{"type":"string"},"title":"Graph Properties Changed","type":"array"},"limitations":{"default":["Static impact within this graph only; external consumers are not inspected.","Affected outputs describe changed definitions, not a selection of outputs to execute. A full run writes every configured output, including unchanged outputs.","Declared write behavior is not proof that this graph can execute. Runtime validation and provider checks still apply; the current graph runtime supports replace targets.","Saving a graph does not write or delete warehouse data."],"items":{"type":"string"},"title":"Limitations","type":"array"},"node_changes":{"items":{"$ref":"#/$defs/GraphNodeChange"},"title":"Node Changes","type":"array"},"outputs":{"items":{"$ref":"#/$defs/GraphOutputImpact"},"title":"Outputs","type":"array"},"run_output_ids":{"items":{"type":"string"},"title":"Run Output Ids","type":"array"}},"required":["baseline_content_sha256","candidate_content_sha256","graph_properties_changed","node_changes","connection_changes","affected_outputs","outputs","run_output_ids"],"title":"GraphChangePreviewResponse","type":"object"} diff --git a/src/dander/control/contracts/v1/schemas/run-explanation.schema.json b/src/dander/control/contracts/v1/schemas/run-explanation.schema.json new file mode 100644 index 00000000..a8c42979 --- /dev/null +++ b/src/dander/control/contracts/v1/schemas/run-explanation.schema.json @@ -0,0 +1 @@ +{"$defs":{"RunActionKind":{"enum":["refresh","logs","replay","cancel","inspect_outputs","review_configuration","none"],"type":"string"},"RunExplanationAction":{"additionalProperties":false,"description":"One suggested next step, without performing a run mutation.","properties":{"kind":{"$ref":"#/$defs/RunActionKind"},"label":{"title":"Label","type":"string"},"reason":{"title":"Reason","type":"string"}},"required":["kind","label","reason"],"title":"RunExplanationAction","type":"object"},"RunState":{"enum":["queued","running","succeeded","failed","canceling","canceled","retrying"],"title":"RunState","type":"string"}},"$id":"urn:dander:control:contracts:v1:run-explanation","$schema":"https://json-schema.org/draft/2020-12/schema","additionalProperties":false,"description":"A bounded explanation derived only from the current normalized run status.","properties":{"available_actions":{"items":{"$ref":"#/$defs/RunActionKind"},"maxItems":7,"title":"Available Actions","type":"array"},"caveats":{"items":{"type":"string"},"maxItems":8,"title":"Caveats","type":"array"},"details":{"items":{"type":"string"},"maxItems":8,"title":"Details","type":"array"},"next_action":{"$ref":"#/$defs/RunExplanationAction"},"results_available":{"title":"Results Available","type":"boolean"},"run_id":{"title":"Run Id","type":"string"},"state":{"$ref":"#/$defs/RunState"},"summary":{"title":"Summary","type":"string"}},"required":["run_id","state","results_available","summary","next_action"],"title":"RunExplanationResponse","type":"object"} diff --git a/src/dander/control/http.py b/src/dander/control/http.py index 26bf0a57..51bc1b4d 100644 --- a/src/dander/control/http.py +++ b/src/dander/control/http.py @@ -554,6 +554,22 @@ def validate_graph(project: str, graph: str, if_match: _IF_MATCH_HEADER) -> obje project, graph, expected_revision=decode_revision_etag(if_match) ) + @app.post( + "/v1/projects/{project}/graphs/{graph}/change-preview", + dependencies=auth_dependencies(ControlCapability.VALIDATE_PREVIEW), + ) + async def preview_graph_changes( + project: str, graph: str, request: Request, if_match: _IF_MATCH_HEADER + ) -> object: + candidate = await read_graph_document(request) + return await run_in_threadpool( + application.preview_graph_changes, + project, + graph, + candidate, + expected_revision=decode_revision_etag(if_match), + ) + @app.post( "/v1/projects/{project}/graphs/{graph}/deployment-preview", dependencies=auth_dependencies(ControlCapability.VALIDATE_PREVIEW), @@ -610,6 +626,13 @@ def list_runs( def get_run(run_id: str) -> object: return application.get_run(RunAddress(run_id)) + @app.get( + "/v1/runs/{run_id}/explanation", + dependencies=auth_dependencies(ControlCapability.READ), + ) + def explain_run(run_id: str) -> object: + return application.explain_run(RunAddress(run_id)) + @app.get( "/v1/runs/{run_id}/logs", dependencies=auth_dependencies(ControlCapability.READ), @@ -672,7 +695,7 @@ def _error_response( def _operation_capability(operation: str) -> ControlCapability: if operation in {"graph.edit"}: return ControlCapability.EDIT - if operation in {"graph.validate", "deployment.preview"}: + if operation in {"graph.validate", "graph.change-preview", "deployment.preview"}: return ControlCapability.VALIDATE_PREVIEW if operation in {"run.start", "run.cancel", "run.replay"}: return ControlCapability.RUN diff --git a/src/dander/control/models.py b/src/dander/control/models.py index 753fe5da..4083b395 100644 --- a/src/dander/control/models.py +++ b/src/dander/control/models.py @@ -797,9 +797,11 @@ class CapabilitiesResponse(ControlModel): "graph.edit", "graph.delete", "graph.validate", + "graph.change-preview", "deployment.preview", "run.start", "run.read", + "run.explain", "run.logs", "run.cancel", "run.replay", diff --git a/src/dander/control/run_explanation.py b/src/dander/control/run_explanation.py new file mode 100644 index 00000000..190c29cf --- /dev/null +++ b/src/dander/control/run_explanation.py @@ -0,0 +1,194 @@ +"""Readable, deterministic explanations of the evidence in a Control run status.""" + +from __future__ import annotations + +from typing import Literal + +from pydantic import Field + +from dander.control.models import ControlModel, RunState, RunStatusResponse + +type RunActionKind = Literal[ + "refresh", "logs", "replay", "cancel", "inspect_outputs", "review_configuration", "none" +] + + +class RunExplanationAction(ControlModel): + """One suggested next step, without performing a run mutation.""" + + kind: RunActionKind + label: str + reason: str + + +class RunExplanationResponse(ControlModel): + """A bounded explanation derived only from the current normalized run status.""" + + run_id: str + state: RunState + results_available: bool + summary: str + details: tuple[str, ...] = Field(default_factory=tuple, max_length=8) + caveats: tuple[str, ...] = Field(default_factory=tuple, max_length=8) + next_action: RunExplanationAction + available_actions: tuple[RunActionKind, ...] = Field(default_factory=tuple, max_length=7) + + +_ACTIVE_STATES = frozenset( + {RunState.QUEUED, RunState.RUNNING, RunState.RETRYING, RunState.CANCELING} +) +_STATE_SUMMARIES = { + RunState.QUEUED: "The run is queued; no completed outcome is available yet.", + RunState.RUNNING: "The run is active; its outcome is not confirmed yet.", + RunState.RETRYING: "The run is retrying; its final outcome is not confirmed yet.", + RunState.CANCELING: "Cancellation was requested; the run has not stopped yet.", + RunState.CANCELED: "The run was canceled.", + RunState.FAILED: "The run failed; the status does not identify the underlying cause.", +} +# Stages and failure summaries can contain provider-supplied free text. Only known stage +# semantics are translated here; raw logs, configuration and arbitrary diagnostics stay out. +_STAGE_DETAILS = { + "submitted": "The execution request has been submitted.", + "starting": "The execution is starting.", + "pending": "The execution is waiting to start.", + "reconciling": "Control is checking the execution's latest state.", + "pending_redrive": "The execution is awaiting a retry decision.", +} + + +def explain_run(status: RunStatusResponse) -> RunExplanationResponse: + """Explain known outcomes without treating missing counters as measured zeros. + + Control may learn that an execution succeeded before it collects the runtime's result + summary. ``result_schema`` is the evidence that counters were actually collected; the + transport's default zeros alone never establish that no rows were processed. + """ + results_available = status.result_schema is not None + details: list[str] = [] + caveats: list[str] = [] + summary = _summary(status, results_available=results_available) + + if status.state in _ACTIVE_STATES and status.stage in _STAGE_DETAILS: + details.append(_STAGE_DETAILS[status.stage]) + if results_available: + details.extend( + ( + f"Recorded extraction: {status.extracted:,} rows " + f"from {status.endpoints:,} endpoints.", + f"Recorded output: {status.affected:,} affected rows and {status.models:,} models.", + f"Recorded validation and catalog: {status.assertions:,} assertions evaluated " + f"and {status.assets:,} assets.", + ) + ) + caveats.append( + "Recorded counts describe this run; they are not a before/after data comparison." + ) + if status.telemetry is not None: + telemetry = status.telemetry + details.append( + f"Recorded runtime: {telemetry.duration_ms:,} ms, " + f"{telemetry.operation_count:,} operations and {telemetry.retry_count:,} retries." + ) + if telemetry.bytes_processed > 0: + details.append(f"Recorded processing: {telemetry.bytes_processed:,} bytes.") + caveats.append( + "Telemetry covers reported operations; zero or omitted counters do not prove " + "that every provider operation was measured. Actual billed cost is not collected." + ) + else: + caveats.append("Runtime telemetry has not been collected for this run.") + elif status.state is RunState.SUCCEEDED: + caveats.append( + "The execution reported success, but its result summary has not been collected. " + "Row counts and completed data checks are not yet verified." + ) + else: + caveats.append( + "No completed result summary is available; row counts and data checks are unknown." + ) + + if status.state in {RunState.FAILED, RunState.CANCELED, RunState.CANCELING}: + caveats.append( + "Stopping or failing does not confirm that earlier writes were rolled back. " + "Inspect the outputs before another attempt." + ) + if status.can_replay: + caveats.append( + "Replay starts another execution; it does not undo prior writes or repair only " + "a selected date range." + ) + + next_action = _next_action(status, results_available=results_available) + actions: list[RunActionKind] = ["refresh"] + if status.logs_available: + actions.append("logs") + if status.can_cancel: + actions.append("cancel") + if status.can_replay: + actions.append("replay") + if next_action.kind not in actions and next_action.kind != "none": + actions.append(next_action.kind) + return RunExplanationResponse( + run_id=status.run_id, + state=status.state, + results_available=results_available, + summary=summary, + details=tuple(details), + caveats=tuple(caveats), + next_action=next_action, + available_actions=tuple(actions), + ) + + +def _summary(status: RunStatusResponse, *, results_available: bool) -> str: + if status.state is RunState.SUCCEEDED: + if not results_available: + return "The execution succeeded; awaiting its result measurements." + if status.skipped: + return "The execution completed, but the pipeline reported that its work was skipped." + return "The run succeeded and its result measurements are available." + if status.state is RunState.FAILED and status.failure_code == "launcher_deadline_exceeded": + return "The execution exceeded its configured time limit." + return _STATE_SUMMARIES[status.state] + + +def _next_action(status: RunStatusResponse, *, results_available: bool) -> RunExplanationAction: + if status.state in _ACTIVE_STATES or ( + status.state is RunState.SUCCEEDED and not results_available + ): + return RunExplanationAction( + kind="refresh", + label="Refresh run status", + reason="Wait for confirmed status and result measurements before starting another run.", + ) + if status.state is RunState.FAILED: + if status.logs_available: + return RunExplanationAction( + kind="logs", + label="Inspect execution logs", + reason="Find the reported error and check for partial writes " + "before deciding to replay.", + ) + return RunExplanationAction( + kind="review_configuration", + label="Review the execution and configuration", + reason="Logs are not available through Control; inspect the execution and its outputs " + "before deciding whether another attempt is appropriate.", + ) + if status.state is RunState.CANCELED: + return RunExplanationAction( + kind="inspect_outputs", + label="Inspect any partial outputs", + reason="Cancellation does not establish that prior writes were undone.", + ) + if status.skipped: + return RunExplanationAction( + kind="logs" if status.logs_available else "review_configuration", + label="Review why the work was skipped", + reason="The result records a skip but does not include its reason.", + ) + return RunExplanationAction( + kind="inspect_outputs", + label="Review the pipeline outputs", + reason="Confirm that the recorded output matches the result you expected.", + ) diff --git a/tests/control/test_change_preview.py b/tests/control/test_change_preview.py new file mode 100644 index 00000000..12d7a0bf --- /dev/null +++ b/tests/control/test_change_preview.py @@ -0,0 +1,241 @@ +"""Static change previews explain impact without saving graphs or reading warehouse rows.""" + +from __future__ import annotations + +from copy import deepcopy +from typing import Any + +import pytest + +from dander.control.change_preview import compare_graph_changes +from dander.control.graph_store import GraphStoreDocumentError, canonicalize_graph_document +from dander.control.models import PipelineGraphDocument + + +def _graph() -> dict[str, Any]: + def node(node_id: str, kind: str) -> dict[str, Any]: + config: dict[str, Any] = {} + if kind == "source": + config = {"connector": "records", "endpoint": node_id} + elif kind == "target": + config = { + "writer": { + "write_mode": "replace", + "destination": {"dataset": "analytics", "table": node_id}, + } + } + return { + "id": node_id, + "name": node_id, + "type": kind, + "config": config, + "fields": [{"name": "id", "type": "STRING"}], + } + + return { + "name": "example", + "nodes": [ + node("source_a", "source"), + node("transform", "transform"), + node("output_a", "target"), + node("source_b", "source"), + node("output_b", "target"), + ], + "edges": [ + {"from": "source_a", "to": "transform"}, + {"from": "transform", "to": "output_a"}, + {"from": "source_b", "to": "output_b"}, + ], + } + + +def _document(payload: dict[str, Any]) -> PipelineGraphDocument: + return PipelineGraphDocument.model_validate(payload) + + +def test_source_edit_marks_only_its_downstream_output_but_full_run_still_writes_both() -> None: + baseline = _graph() + candidate = deepcopy(baseline) + candidate["nodes"][0]["config"]["endpoint"] = "changed_endpoint" + candidate["nodes"][0]["config"]["private_note"] = "do-not-copy-this-configuration" + + result = compare_graph_changes(_document(baseline), _document(candidate)) + + assert result.baseline_content_sha256 == canonicalize_graph_document(baseline).content_sha256 + assert result.candidate_content_sha256 == canonicalize_graph_document(candidate).content_sha256 + assert result.node_changes[0].node_id == "source_a" + assert result.node_changes[0].change == "modified" + assert result.node_changes[0].changed_properties == ("config",) + assert result.affected_outputs == ("output_a",) + assert result.run_output_ids == ("output_a", "output_b") + assert not result.outputs[1].affected + assert "do-not-copy-this-configuration" not in result.model_dump_json() + assert baseline == _graph() + + +@pytest.mark.parametrize("remove_node", [False, True]) +def test_disconnecting_a_branch_retains_its_previous_downstream_impact(remove_node: bool) -> None: + baseline = _graph() + candidate = deepcopy(baseline) + candidate["edges"] = [edge for edge in candidate["edges"] if edge["from"] != "source_a"] + if remove_node: + candidate["nodes"] = [node for node in candidate["nodes"] if node["id"] != "source_a"] + + result = compare_graph_changes(_document(baseline), _document(candidate)) + + assert result.affected_outputs == ("output_a",) + assert result.connection_changes[0].change == "removed" + assert result.connection_changes[0].target == "transform" + if remove_node: + assert result.node_changes[0].change == "removed" + + +def test_layout_only_edit_changes_identity_without_claiming_data_impact() -> None: + baseline = _graph() + candidate = deepcopy(baseline) + candidate["nodes"][0]["visual"] = {"position": {"x": 20, "y": 30}} + + result = compare_graph_changes(_document(baseline), _document(candidate)) + + assert result.baseline_content_sha256 != result.candidate_content_sha256 + assert result.node_changes[0].change == "layout_only" + assert not result.node_changes[0].affects_data + assert result.affected_outputs == () + assert result.run_output_ids == ("output_a", "output_b") + + +def test_changed_field_mapping_marks_downstream_output_without_echoing_expression_values() -> None: + baseline = _graph() + baseline["edges"][0]["mappings"] = [{"source": "id", "target": "id"}] + candidate = deepcopy(baseline) + candidate["edges"][0]["mappings"] = [ + { + "source": None, + "target": "id", + "transformation": {"kind": "constant", "constant": "private-config-value"}, + } + ] + + result = compare_graph_changes(_document(baseline), _document(candidate)) + + assert result.node_changes == () + assert result.connection_changes[0].change == "modified" + assert result.affected_outputs == ("output_a",) + assert "private-config-value" not in result.model_dump_json() + + +@pytest.mark.parametrize("schedule_changed", [False, True]) +def test_graph_schedule_change_affects_all_outputs_but_graph_title_does_not( + schedule_changed: bool, +) -> None: + baseline = _graph() + candidate = deepcopy(baseline) + if schedule_changed: + candidate["trigger"] = {"kind": "schedule", "cron": "0 6 * * *"} + else: + candidate["name"] = "A clearer title" + + result = compare_graph_changes(_document(baseline), _document(candidate)) + + assert result.graph_properties_changed == (("trigger",) if schedule_changed else ("name",)) + assert result.affected_outputs == (("output_a", "output_b") if schedule_changed else ()) + + +def test_moving_and_removing_outputs_explains_existing_data_is_retained() -> None: + baseline = _graph() + candidate = deepcopy(baseline) + candidate["nodes"][2]["config"]["writer"]["destination"]["table"] = "new_output" + candidate["nodes"] = [node for node in candidate["nodes"] if node["id"] != "output_b"] + candidate["edges"] = [edge for edge in candidate["edges"] if edge["to"] != "output_b"] + + result = compare_graph_changes(_document(baseline), _document(candidate)) + moved, removed = result.outputs + + assert moved.before is not None and moved.before.writer is not None + assert moved.after is not None and moved.after.writer is not None + assert moved.before.writer.destination.table == "output_a" + assert moved.after.writer.destination.table == "new_output" + assert "previous destination is retained" in moved.effect + assert "Replaces all rows" in moved.effect + assert removed.before is not None + assert removed.after is None + assert "existing warehouse data is not deleted" in removed.effect + assert result.affected_outputs == ("output_a", "output_b") + assert result.run_output_ids == ("output_a",) + + +def test_added_output_reports_declared_schema_and_unknown_cost_without_execution_claim() -> None: + baseline = _graph() + candidate = deepcopy(baseline) + new_target = deepcopy(candidate["nodes"][2]) + new_target["id"] = "output_c" + new_target["name"] = "Another output" + new_target["config"]["writer"]["destination"]["table"] = "output_c" + new_target["config"]["writer"]["destination"]["business_key"] = ["id"] + new_target["config"]["writer"]["write_mode"] = "scd1" + candidate["nodes"].append(new_target) + candidate["edges"].append({"from": "transform", "to": "output_c"}) + + result = compare_graph_changes(_document(baseline), _document(candidate)) + added = result.outputs[2] + + assert added.before is None + assert added.after is not None + assert added.after.field_names == ("id",) + assert added.after.writer is not None + assert added.after.writer.write_mode == "scd1" + assert "New graph output" in added.effect + assert "Updates matching business keys" in added.effect + assert result.affected_outputs == ("output_c",) + assert result.estimates.rows_written is None + assert result.estimates.cost_usd is None + assert any("not proof that this graph can execute" in note for note in result.limitations) + + +def test_target_without_writer_and_reordered_graph_remain_honest_and_deterministic() -> None: + baseline = _graph() + baseline["nodes"][2]["config"] = {} + candidate = deepcopy(baseline) + candidate["nodes"].reverse() + candidate["edges"].reverse() + + result = compare_graph_changes(_document(baseline), _document(candidate)) + + assert result.node_changes == () + assert result.connection_changes == () + assert result.affected_outputs == () + assert result.run_output_ids == ("output_a", "output_b") + assert "no declared writer" in result.outputs[0].effect + + +def test_union_of_old_and_new_edges_can_be_cyclic_without_hanging_preview() -> None: + baseline = _graph() + other_transform = deepcopy(baseline["nodes"][1]) + other_transform["id"] = "other_transform" + baseline["nodes"].append(other_transform) + baseline["edges"] = [ + {"from": "source_a", "to": "transform"}, + {"from": "transform", "to": "other_transform"}, + {"from": "other_transform", "to": "output_a"}, + {"from": "source_b", "to": "output_b"}, + ] + candidate = deepcopy(baseline) + candidate["edges"] = [ + {"from": "source_a", "to": "other_transform"}, + {"from": "other_transform", "to": "transform"}, + {"from": "transform", "to": "output_a"}, + {"from": "source_b", "to": "output_b"}, + ] + + result = compare_graph_changes(_document(baseline), _document(candidate)) + + assert result.affected_outputs == ("output_a",) + + +def test_preview_reuses_graph_store_inline_credential_rejection() -> None: + baseline = _graph() + candidate = deepcopy(baseline) + candidate["nodes"][0]["config"]["password"] = "synthetic-secret" + + with pytest.raises(GraphStoreDocumentError, match="canonical graph contract"): + compare_graph_changes(_document(baseline), _document(candidate)) diff --git a/tests/control/test_hosted_control.py b/tests/control/test_hosted_control.py index 6d377d42..778d731c 100644 --- a/tests/control/test_hosted_control.py +++ b/tests/control/test_hosted_control.py @@ -164,6 +164,7 @@ def test_capabilities_are_honest_and_unwired_operations_fail_closed(client: Test "graph.edit", "graph.delete", "graph.validate", + "graph.change-preview", } created = _create(client, "alpha-graph") preview = client.post( @@ -201,6 +202,47 @@ def test_run_start_rejects_a_body_before_operation_dispatch(client: TestClient) assert response.json()["error"]["code"] == "request_invalid" +def test_change_preview_uses_saved_revision_without_saving_candidate(client: TestClient) -> None: + with client: + created = _create(client, "alpha-graph") + route = "/v1/projects/demo-project/graphs/alpha-graph" + headers = {"If-Match": created.headers["etag"]} + preview = client.post(f"{route}/change-preview", json=GRAPH_TWO, headers=headers) + assert preview.status_code == 200 + result = preview.json() + assert result["baseline_content_sha256"] == created.json()["content_sha256"] + assert result["candidate_content_sha256"] != result["baseline_content_sha256"] + saved = client.get(route) + assert saved.headers["etag"] == created.headers["etag"] + assert saved.json() == created.json() + + assert client.put(route, json=GRAPH_TWO, headers=headers).status_code == 200 + stale = client.post(f"{route}/change-preview", json=GRAPH_TWO, headers=headers) + assert stale.status_code == 412 + missing = client.post(f"{route}/change-preview", json=GRAPH_TWO) + assert missing.status_code == 422 + + +def test_change_preview_reuses_graph_validation_and_size_limits(client: TestClient) -> None: + with client: + created = _create(client, "alpha-graph") + route = "/v1/projects/demo-project/graphs/alpha-graph/change-preview" + headers = {"If-Match": created.headers["etag"]} + invalid = client.post(route, json={**GRAPH, "unexpected": True}, headers=headers) + assert invalid.status_code == 422 + oversized = client.post( + route, + content=b"{}", + headers={ + **headers, + "Content-Type": "application/json", + "Content-Length": str(5 * 1024 * 1024 + 1), + }, + ) + assert oversized.status_code == 413 + assert client.get("/v1/projects/demo-project/graphs/alpha-graph").json() == created.json() + + def test_graph_delete_rejects_a_body_before_mutation(client: TestClient) -> None: with client: created = _create(client, "alpha-graph") @@ -375,6 +417,12 @@ def test_normalized_lifecycle_receives_decoded_revision_and_explicit_idempotency assert selected.status_code == 202 assert lifecycle.starts[-1].environment == "gcp" assert client.get("/v1/runs").json()["items"][0]["run_id"] == "run-one" + explanation = client.get("/v1/runs/run-one/explanation") + assert explanation.status_code == 200 + assert explanation.json()["run_id"] == "run-one" + assert explanation.json()["state"] == "queued" + assert "run.explain" in client.get("/v1/capabilities").json()["operations"] + assert lifecycle.mutations == [] assert client.get("/v1/runs/run-one/logs?limit=25").status_code == 200 assert ( client.post( diff --git a/tests/control/test_oidc_auth.py b/tests/control/test_oidc_auth.py index 13ba8d89..228fcc56 100644 --- a/tests/control/test_oidc_auth.py +++ b/tests/control/test_oidc_auth.py @@ -270,6 +270,13 @@ def test_each_route_capability_is_enforced_server_side() -> None: created = _create(client, "editor") assert created.status_code == 201 + for role, expected_status in (("viewer", 403), ("editor", 200)): + preview = client.post( + "/v1/projects/demo-project/graphs/alpha-graph/change-preview", + json=created.json()["document"], + headers={**_authorization(role), "If-Match": created.headers["etag"]}, + ) + assert preview.status_code == expected_status assert ( client.post( "/v1/projects/demo-project/graphs/alpha-graph/validate", @@ -315,12 +322,18 @@ def test_capabilities_are_filtered_for_the_authenticated_role() -> None: editor = client.get("/v1/capabilities", headers=_authorization("editor")).json() admin = client.get("/v1/capabilities", headers=_authorization("admin")).json() assert viewer["operations"] == ["graph.read"] - assert set(editor["operations"]) == {"graph.read", "graph.edit", "graph.validate"} + assert set(editor["operations"]) == { + "graph.read", + "graph.edit", + "graph.validate", + "graph.change-preview", + } assert set(admin["operations"]) == { "graph.read", "graph.edit", "graph.delete", "graph.validate", + "graph.change-preview", } diff --git a/tests/control/test_run_explanation.py b/tests/control/test_run_explanation.py new file mode 100644 index 00000000..8a67cd2a --- /dev/null +++ b/tests/control/test_run_explanation.py @@ -0,0 +1,203 @@ +"""Run explanations distinguish observed outcomes from missing measurements.""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from dander.control.models import RunState, RunStatusResponse, RunTelemetrySummary +from dander.control.run_explanation import RunExplanationResponse, explain_run + +_RESULT_SCHEMA = "io.dander.control.execution-result-summary/v1" + + +def test_success_explains_collected_counts_without_claiming_a_data_diff() -> None: + result = explain_run( + RunStatusResponse( + run_id="successful", + state=RunState.SUCCEEDED, + result_schema=_RESULT_SCHEMA, + endpoints=2, + extracted=1234, + affected=27, + models=1, + assertions=3, + assets=1, + ) + ) + + assert result.results_available + assert any("1,234 rows from 2 endpoints" in detail for detail in result.details) + assert any("27 affected rows" in detail for detail in result.details) + assert any("3 assertions evaluated" in detail for detail in result.details) + assert any("not a before/after data comparison" in caveat for caveat in result.caveats) + assert result.next_action.kind == "inspect_outputs" + assert "replay" not in result.available_actions + + +def test_success_without_collected_results_does_not_turn_defaults_into_measured_zeros() -> None: + result = explain_run(RunStatusResponse(run_id="awaiting", state=RunState.SUCCEEDED)) + + assert not result.results_available + assert result.details == () + assert "awaiting" in result.summary + assert any("not yet verified" in caveat for caveat in result.caveats) + assert result.next_action.kind == "refresh" + assert "0 rows" not in result.model_dump_json() + + +def test_counters_without_result_contract_do_not_establish_completed_work() -> None: + result = explain_run( + RunStatusResponse( + run_id="uncollected", + state=RunState.SUCCEEDED, + extracted=900, + affected=500, + skipped=True, + ) + ) + + assert not result.results_available + assert result.details == () + assert "skipped" not in result.summary + assert result.next_action.kind == "refresh" + + +@pytest.mark.parametrize("logs_available", [False, True]) +def test_skipped_work_does_not_invent_why_it_was_skipped(logs_available: bool) -> None: + result = explain_run( + RunStatusResponse( + run_id="skipped", + state=RunState.SUCCEEDED, + result_schema=_RESULT_SCHEMA, + skipped=True, + logs_available=logs_available, + ) + ) + + assert result.results_available + assert "skipped" in result.summary + assert "0 rows" in " ".join(result.details) + assert "does not include its reason" in result.next_action.reason + assert result.next_action.kind == ("logs" if logs_available else "review_configuration") + + +@pytest.mark.parametrize( + "state", [RunState.QUEUED, RunState.RUNNING, RunState.RETRYING, RunState.CANCELING] +) +def test_active_runs_recommend_observation_and_respect_cancel_availability(state: RunState) -> None: + result = explain_run( + RunStatusResponse( + run_id="active", + state=state, + stage="reconciling", + can_cancel=state is not RunState.CANCELING, + ) + ) + + assert result.next_action.kind == "refresh" + assert "replay" not in result.available_actions + assert ("cancel" in result.available_actions) == (state is not RunState.CANCELING) + assert any("checking" in detail for detail in result.details) + assert not result.results_available + + +def test_failed_run_points_to_available_logs_without_echoing_unsafe_diagnostics() -> None: + result = explain_run( + RunStatusResponse( + run_id="failed", + state=RunState.FAILED, + stage="private-stage-with-token", + failure_code="unrecognized-private-error", + failure_summary="provider-password=private-diagnostic", + logs_available=True, + can_replay=True, + ) + ) + + assert result.next_action.kind == "logs" + assert "replay" in result.available_actions + assert "cancel" not in result.available_actions + assert "underlying cause" in result.summary + assert "private" not in result.model_dump_json() + assert any("rolled back" in caveat for caveat in result.caveats) + assert any("does not undo prior writes" in caveat for caveat in result.caveats) + + +def test_failed_run_without_logs_never_suggests_unavailable_logs_or_replay() -> None: + result = explain_run(RunStatusResponse(run_id="failed", state=RunState.FAILED)) + + assert result.next_action.kind == "review_configuration" + assert "logs" not in result.available_actions + assert "replay" not in result.available_actions + assert "unknown" in " ".join(result.caveats) + + +def test_known_timeout_does_not_invent_the_cause_of_slow_execution() -> None: + result = explain_run( + RunStatusResponse( + run_id="timeout", + state=RunState.FAILED, + failure_code="launcher_deadline_exceeded", + logs_available=True, + ) + ) + + assert "time limit" in result.summary + assert result.next_action.kind == "logs" + assert "memory" not in result.model_dump_json() + + +def test_canceled_run_warns_about_existing_writes_before_replay() -> None: + result = explain_run( + RunStatusResponse(run_id="canceled", state=RunState.CANCELED, can_replay=True) + ) + + assert result.next_action.kind == "inspect_outputs" + assert "replay" in result.available_actions + assert "partial" in result.next_action.label + assert any("rolled back" in caveat for caveat in result.caveats) + + +def test_telemetry_reports_observed_runtime_without_claiming_zero_cost_or_full_coverage() -> None: + result = explain_run( + RunStatusResponse( + run_id="telemetry", + state=RunState.SUCCEEDED, + result_schema=_RESULT_SCHEMA, + telemetry=RunTelemetrySummary( + duration_ms=2500, + operation_count=3, + retry_count=1, + rows_read=0, + rows_written=0, + rows_affected=0, + bytes_read=0, + bytes_written=0, + bytes_processed=2048, + bytes_billed=0, + queue_duration_ms=0, + execution_duration_ms=0, + spill_bytes=0, + ), + ) + ) + + assert any("2,500 ms" in detail and "1 retries" in detail for detail in result.details) + assert any("2,048 bytes" in detail for detail in result.details) + assert any("Actual billed cost is not collected" in caveat for caveat in result.caveats) + assert any("every provider operation" in caveat for caveat in result.caveats) + + +def test_explanation_contract_rejects_extra_fields_at_each_object_boundary() -> None: + result = explain_run(RunStatusResponse(run_id="closed", state=RunState.QUEUED)) + payload = result.model_dump(mode="json") + with pytest.raises(ValidationError, match="extra_forbidden"): + RunExplanationResponse.model_validate({**payload, "raw_config": {}}) + with pytest.raises(ValidationError, match="extra_forbidden"): + RunExplanationResponse.model_validate( + { + **payload, + "next_action": {**result.next_action.model_dump(), "command": "run-anything"}, + } + ) diff --git a/tickets/DANDER-286-guided-change-and-run-explanations.md b/tickets/DANDER-286-guided-change-and-run-explanations.md new file mode 100644 index 00000000..e44fbd7b --- /dev/null +++ b/tickets/DANDER-286-guided-change-and-run-explanations.md @@ -0,0 +1,35 @@ +--- +id: DANDER-286 +title: Explain unsaved graph changes and recorded run outcomes +status: in-code +component: python +depends_on: [] +created: 2026-10-01 +--- + +## Context + +The guided connect, preview, run, and understand journey needs a read-only data-impact preview +distinct from infrastructure deployment, and a readable account of actual run evidence. + +## Acceptance Criteria + +- [x] Compare an unsaved candidate against the exact saved graph revision without mutation. +- [x] Explain downstream effects, all full-run outputs, destination moves, and output removal. +- [x] Keep row/cost estimates unknown and distinguish static impact from execution eligibility. +- [x] Explain normalized run status without invented causes or missing-as-zero measurements. +- [x] Reuse existing authentication, bounded graph parsing, lifecycle, and generated contracts. +- [x] Verify stale revision, authorization, no-save behavior, and deterministic contract generation. + +## Implementation Notes + +Pure modules own comparisons and explanations. Control exposes `/change-preview` for a saved +graph plus candidate and `/explanation` for a run. Published RC19 clients remain paired with their +published producer; the new source bundle requires a newly generated client from its immutable +release artifact. This ticket does not complete the overall guided journey: Druff integration and +real date-scoped output repair remain follow-on work. + +## Verification + +Focused comparison, explanation, HTTP, OIDC, and contract tests passed. Canonical strict typing +passed across 513 files. Ruff and generated contract drift checks passed. From 80aeae459ee5544e44d565b0774fb1e369c0c994 Mon Sep 17 00:00:00 2001 From: Harrison O'Connor-Hoover Date: Thu, 1 Oct 2026 05:56:02 -0400 Subject: [PATCH 2/2] Regenerate Kubernetes bootstrap fixture for Control contract --- tests/fixtures/kubernetes-control-values.yaml | 44 ++++++++++++------- 1 file changed, 28 insertions(+), 16 deletions(-) diff --git a/tests/fixtures/kubernetes-control-values.yaml b/tests/fixtures/kubernetes-control-values.yaml index 371f9e3d..e3572a8e 100644 --- a/tests/fixtures/kubernetes-control-values.yaml +++ b/tests/fixtures/kubernetes-control-values.yaml @@ -8,22 +8,28 @@ serviceAccount: control: port: 8770 command: - - control - - serve - - --host - - 0.0.0.0 - - --port - - "8770" - - --oidc-config - - /etc/dander/control-oidc.json - - --graph-store-config - - /etc/dander/control-graph-store.json + - control + - serve + - --host + - 0.0.0.0 + - --port + - '8770' + - --oidc-config + - /etc/dander/control-oidc.json + - --graph-store-config + - /etc/dander/control-graph-store.json oidcConfig: '{"allowed_algorithms":["RS256"],"allowed_origins":["https://dander.127.0.0.1.sslip.io:8443"],"api_audience":"dander-kubernetes-control","api_url":"https://dander.127.0.0.1.sslip.io:8443","clock_skew_seconds":30,"email_allowlist":[],"email_claim":"email","group_allowlist":[],"group_claim":"groups","issuer":"https://issuer.127.0.0.1.sslip.io:8443/default","jwks_cache_seconds":300.0,"jwks_max_keys":32,"jwks_max_response_bytes":262144,"jwks_refresh_cooldown_seconds":30.0,"jwks_timeout_seconds":5.0,"jwks_uri":"https://issuer.127.0.0.1.sslip.io:8443/default/.well-known/jwks.json","logout_uri":"https://dander.127.0.0.1.sslip.io:8443/signed-out","public_client_id":"druff-kubernetes-spa","redirect_uri":"https://dander.127.0.0.1.sslip.io:8443/auth/callback","role_claim":"roles","subject_allowlist":[]}' graphStoreConfig: '{"kind":"local","root":"/var/lib/dander/control"}' rolloutDigest: 1fa434c78d6add1ed3f132b42d5398260e0ab61874221f5a75aa6be1e7166474 resources: - requests: {cpu: 500m, memory: 512Mi, ephemeral-storage: 128Mi} - limits: {cpu: 500m, memory: 512Mi, ephemeral-storage: 128Mi} + requests: + cpu: 500m + memory: 512Mi + ephemeral-storage: 128Mi + limits: + cpu: 500m + memory: 512Mi + ephemeral-storage: 128Mi probes: livenessPath: /healthz readinessPath: /readyz @@ -31,11 +37,17 @@ control: image: 127.0.0.1:5001/dander@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa druff: port: 8080 - bootstrap: '{"api_audience":"dander-kubernetes-control","api_url":"https://dander.127.0.0.1.sslip.io:8443","compatibility":{"maximum_druff_contract":"1.x","minimum_druff_contract":"1.0.0"},"contract":{"id":"io.dander.control.contracts/v1","sha256":"bddaca593832c1145409dfb1745cf67d0bd0d62531061113d06055b71f5fa5b1"},"issuer":"https://issuer.127.0.0.1.sslip.io:8443/default","logout_uri":"https://dander.127.0.0.1.sslip.io:8443/signed-out","public_client_id":"druff-kubernetes-spa","redirect_uri":"https://dander.127.0.0.1.sslip.io:8443/auth/callback","schema_version":1}' - rolloutDigest: 87b0e38d3639ec3e604f76df976f3978c55ad7b685126d6f835cd98e84f1cdfd + bootstrap: '{"api_audience":"dander-kubernetes-control","api_url":"https://dander.127.0.0.1.sslip.io:8443","compatibility":{"maximum_druff_contract":"1.x","minimum_druff_contract":"1.0.0"},"contract":{"id":"io.dander.control.contracts/v1","sha256":"03fdac1a19a98f10f1af6fa7821b705291833f88a23e723300abace660130d5b"},"issuer":"https://issuer.127.0.0.1.sslip.io:8443/default","logout_uri":"https://dander.127.0.0.1.sslip.io:8443/signed-out","public_client_id":"druff-kubernetes-spa","redirect_uri":"https://dander.127.0.0.1.sslip.io:8443/auth/callback","schema_version":1}' + rolloutDigest: aa5f974008827a19774aa39b0ebf9bc953f947012db47ed3e8e72b65ad3f90b1 resources: - requests: {cpu: 100m, memory: 128Mi, ephemeral-storage: 64Mi} - limits: {cpu: 500m, memory: 256Mi, ephemeral-storage: 128Mi} + requests: + cpu: 100m + memory: 128Mi + ephemeral-storage: 64Mi + limits: + cpu: 500m + memory: 256Mi + ephemeral-storage: 128Mi image: 127.0.0.1:5001/druff@sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc graphStore: accessMode: ReadWriteOnce