Skip to content

feat: add public lightweight dataops run #103

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zhouning merged 2 commits into from
Aug 17, 2026
Merged

feat: add public lightweight dataops run #103

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5d7ef60
feat: add public lightweight dataops run
zhouning Aug 17, 2026
ec056cd
test: recognize synchronous success migration
zhouning Aug 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .github/workflows/cd-staging.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -173,6 +173,8 @@ jobs:
data_agent/test_chongqing_protected_admission_workflow.py \
data_agent/test_public_source_landing.py \
data_agent/test_public_source_landing_postgres.py \
data_agent/test_public_dataops_run.py \
data_agent/test_public_dataops_run_postgres.py \
data_agent/test_dolphinscheduler_adapter.py \
data_agent/test_dolphinscheduler_command_consumer.py \
data_agent/test_dolphinscheduler_command_worker.py \
Expand Down
2 changes: 2 additions & 0 deletions .github/workflows/ci.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -182,6 +182,8 @@ jobs:
data_agent/test_chongqing_protected_admission_workflow.py \
data_agent/test_public_source_landing.py \
data_agent/test_public_source_landing_postgres.py \
data_agent/test_public_dataops_run.py \
data_agent/test_public_dataops_run_postgres.py \
data_agent/test_dolphinscheduler_adapter.py \
data_agent/test_dolphinscheduler_command_consumer.py \
data_agent/test_dolphinscheduler_command_worker.py \
Expand Down
226 changes: 226 additions & 0 deletions data_agent/migrations/099_synchronous_success_profile.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,226 @@
-- 099: Evidence-gated success for the synchronous public DataOps profile.
--
-- Migration 096 remains checksum-frozen and continues to own the
-- DolphinScheduler verdict. This separate entry point accepts only the local
-- inline observation profile while preserving the same output, quality,
-- lineage, actor, tenant, state, fingerprint, and replay checks.

CREATE OR REPLACE FUNCTION gda_control.finalize_synchronous_platform_run_success(
p_tenant_id TEXT,
p_run_id UUID,
p_expected_state_version INTEGER,
p_actor_subject TEXT,
p_reason TEXT,
p_details JSONB
)
RETURNS INTEGER
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, gda_control
SET row_security = on
AS $$
DECLARE
v_run gda_control.platform_run%ROWTYPE;
v_event gda_control.platform_run_event%ROWTYPE;
v_observation_id UUID;
v_output_artifact_id UUID;
v_quality_result_id UUID;
v_lineage_event_id UUID;
v_expected_evidence_sha256 TEXT;
v_output gda_control.artifact%ROWTYPE;
v_quality gda_control.quality_result%ROWTYPE;
BEGIN
IF gda_control.current_tenant() IS NULL
OR p_tenant_id IS DISTINCT FROM gda_control.current_tenant() THEN
RAISE EXCEPTION 'platform run tenant context is missing or mismatched'
USING ERRCODE = '42501';
END IF;
IF NULLIF(btrim(p_actor_subject), '') IS NULL
OR NULLIF(btrim(p_reason), '') IS NULL THEN
RAISE EXCEPTION 'finalization actor and reason are required'
USING ERRCODE = '22023';
END IF;
IF jsonb_typeof(p_details) <> 'object'
OR p_details->>'schema' <> 'gda.run_success_evidence.v1'
OR p_details->>'tenant_id' IS DISTINCT FROM p_tenant_id
OR p_details->>'run_id' IS DISTINCT FROM p_run_id::text
OR COALESCE(p_details->>'evidence_sha256', '')
!~ '^[0-9a-f]{64}$' THEN
RAISE EXCEPTION 'success evidence envelope is invalid'
USING ERRCODE = '22023';
END IF;

SELECT * INTO v_run
FROM gda_control.platform_run
WHERE tenant_id = p_tenant_id AND run_id = p_run_id
FOR UPDATE;
IF NOT FOUND THEN
RAISE EXCEPTION 'platform run % not found', p_run_id
USING ERRCODE = 'P0002';
END IF;

IF v_run.orchestration_class <> 'synchronous' THEN
RAISE EXCEPTION 'synchronous success profile requires synchronous Run'
USING ERRCODE = '23514';
END IF;

IF v_run.status = 'succeeded' THEN
SELECT * INTO v_event
FROM gda_control.platform_run_event
WHERE tenant_id = p_tenant_id
AND run_id = p_run_id
AND sequence_no = v_run.state_version;
IF FOUND
AND v_event.to_status = 'succeeded'
AND v_event.actor_subject = p_actor_subject
AND v_event.reason = p_reason
AND v_event.details = p_details THEN
RETURN v_run.state_version;
END IF;
RAISE EXCEPTION 'successful Run has a different terminal verdict'
USING ERRCODE = '40001';
END IF;

IF v_run.state_version <> p_expected_state_version THEN
RAISE EXCEPTION 'platform run state version conflict: expected %, actual %',
p_expected_state_version, v_run.state_version
USING ERRCODE = '40001';
END IF;
IF v_run.status <> 'running' THEN
RAISE EXCEPTION 'synchronous success finalization requires running Run'
USING ERRCODE = '23514';
END IF;
IF p_actor_subject IS DISTINCT FROM concat(
v_run.subject_context->>'subject_type',
':',
v_run.subject_context->>'subject_id'
) THEN
RAISE EXCEPTION 'finalization actor does not match Run workload'
USING ERRCODE = '42501';
END IF;

BEGIN
v_observation_id := (p_details->>'attempt_observation_id')::uuid;
v_output_artifact_id := (p_details->>'output_artifact_id')::uuid;
v_quality_result_id := (p_details->>'quality_result_id')::uuid;
v_lineage_event_id := (p_details->>'lineage_event_id')::uuid;
EXCEPTION WHEN invalid_text_representation OR null_value_not_allowed THEN
RAISE EXCEPTION 'success evidence identifiers must be UUIDs'
USING ERRCODE = '22023';
END;

v_expected_evidence_sha256 := encode(
sha256(
convert_to(
'{"attempt_observation_id":'
|| to_json(v_observation_id::text)::text
|| ',"lineage_event_id":'
|| to_json(v_lineage_event_id::text)::text
|| ',"output_artifact_id":'
|| to_json(v_output_artifact_id::text)::text
|| ',"quality_result_id":'
|| to_json(v_quality_result_id::text)::text
|| ',"run_id":'
|| to_json(p_run_id::text)::text
|| ',"tenant_id":'
|| to_json(p_tenant_id)::text
|| '}',
'UTF8'
)
),
'hex'
);
IF p_details->>'evidence_sha256'
IS DISTINCT FROM v_expected_evidence_sha256 THEN
RAISE EXCEPTION 'success evidence fingerprint does not match its bindings'
USING ERRCODE = '22023';
END IF;

PERFORM 1
FROM gda_control.framework_attempt_observation
WHERE tenant_id = p_tenant_id
AND observation_id = v_observation_id
AND run_id = p_run_id
AND framework_kind = 'legacy'
AND lower(observed_state) = 'success'
AND evidence->>'schema' = 'gda.public_dataops_attempt.v1'
AND evidence->>'execution_mode' = 'local_inline';
IF NOT FOUND THEN
RAISE EXCEPTION 'local inline success observation was not found'
USING ERRCODE = '23514';
END IF;

SELECT artifact.* INTO v_output
FROM gda_control.artifact AS artifact
JOIN gda_control.resource_version AS version
ON version.tenant_id = artifact.tenant_id
AND version.resource_version_id = artifact.resource_version_id
AND version.content_sha256 = artifact.content_sha256
WHERE artifact.tenant_id = p_tenant_id
AND artifact.artifact_id = v_output_artifact_id
AND artifact.run_id = p_run_id
AND artifact.artifact_role = 'output';
IF NOT FOUND THEN
RAISE EXCEPTION 'content-bound output Artifact was not found'
USING ERRCODE = '23514';
END IF;

SELECT quality.* INTO v_quality
FROM gda_control.quality_result AS quality
JOIN gda_control.artifact AS evidence
ON evidence.tenant_id = quality.tenant_id
AND evidence.artifact_id = quality.evidence_artifact_id
WHERE quality.tenant_id = p_tenant_id
AND quality.quality_result_id = v_quality_result_id
AND quality.run_id = p_run_id
AND quality.resource_version_id = v_output.resource_version_id
AND quality.verdict = 'passed'
AND quality.evaluated_by <> p_actor_subject
AND quality.evaluated_at >= v_output.created_at
AND evidence.artifact_role = 'evidence'
AND evidence.run_id = p_run_id
AND evidence.resource_version_id = v_output.resource_version_id
AND evidence.created_by = quality.evaluated_by;
IF NOT FOUND THEN
RAISE EXCEPTION 'independent passed QualityResult was not found'
USING ERRCODE = '23514';
END IF;

PERFORM 1
FROM gda_control.lineage_event AS lineage
WHERE lineage.tenant_id = p_tenant_id
AND lineage.lineage_event_id = v_lineage_event_id
AND lineage.run_id = p_run_id
AND lineage.definition_version_id = v_run.definition_version_id
AND lineage.artifact_id = v_output_artifact_id
AND lineage.target_resource_version_id = v_output.resource_version_id
AND EXISTS (
SELECT 1
FROM gda_control.platform_run_input_binding AS input
WHERE input.tenant_id = p_tenant_id
AND input.run_id = p_run_id
AND input.resource_version_id = lineage.source_resource_version_id
);
IF NOT FOUND THEN
RAISE EXCEPTION 'input-to-output LineageEvent was not found'
USING ERRCODE = '23514';
END IF;

RETURN gda_control.apply_platform_run_transition(
p_tenant_id,
p_run_id,
p_expected_state_version,
'succeeded',
p_actor_subject,
p_reason,
p_details
);
END;
$$;

REVOKE ALL ON FUNCTION gda_control.finalize_synchronous_platform_run_success(
text, uuid, integer, text, text, jsonb
) FROM PUBLIC;
GRANT EXECUTE ON FUNCTION gda_control.finalize_synchronous_platform_run_success(
text, uuid, integer, text, text, jsonb
) TO gda_control_gateway;
41 changes: 36 additions & 5 deletions data_agent/platform_gateway.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,11 @@
/ "migrations"
/ "096_platform_success_verdict.sql"
)
SYNCHRONOUS_SUCCESS_MIGRATION = (
Path(__file__).resolve().parent
/ "migrations"
/ "099_synchronous_success_profile.sql"
)
METADATA_FABRIC_BINDING_MIGRATION = (
Path(__file__).resolve().parent
/ "migrations"
Expand Down Expand Up @@ -1002,10 +1007,20 @@ def finalize_run_success(
**evidence.model_dump(mode="json"),
}
with self._transaction(evidence.tenant_id) as connection:
run = self._load_run(
connection, evidence.tenant_id, evidence.run_id
)
if run is None:
raise GatewayNotFoundError("PlatformRun was not found")
finalizer = (
"gda_control.finalize_synchronous_platform_run_success"
if run.orchestration_class.value == "synchronous"
else "gda_control.finalize_platform_run_success"
)
connection.execute(
text(
"""
SELECT gda_control.finalize_platform_run_success(
f"""
SELECT {finalizer}(
:tenant_id, :run_id, :expected_state_version,
:actor_subject, :reason, CAST(:details AS jsonb)
)
Expand All @@ -1020,12 +1035,12 @@ def finalize_run_success(
"details": _json(details),
},
).scalar_one()
run = self._load_run(
finalized = self._load_run(
connection, evidence.tenant_id, evidence.run_id
)
if run is None:
if finalized is None:
raise GatewayNotFoundError("PlatformRun was not found")
return run
return finalized

@classmethod
def _reconcile_command(
Expand Down Expand Up @@ -1953,6 +1968,7 @@ def build_gateway_report(
role_migration: Path | None = None,
command_migration: Path | None = None,
success_migration: Path | None = None,
synchronous_success_migration: Path | None = None,
binding_migration: Path | None = None,
lineage_migration: Path | None = None,
gateway_source: Path | None = None,
Expand All @@ -1970,6 +1986,9 @@ def build_gateway_report(
"success_migration": (
success_migration or SUCCESS_VERDICT_MIGRATION
).resolve(),
"synchronous_success_migration": (
synchronous_success_migration or SYNCHRONOUS_SUCCESS_MIGRATION
).resolve(),
"binding_migration": (
binding_migration or METADATA_FABRIC_BINDING_MIGRATION
).resolve(),
Expand Down Expand Up @@ -2037,6 +2056,17 @@ def build_gateway_report(
"GRANT SELECT, INSERT ON gda_control.quality_result",
"finalize_platform_run_success",
),
"synchronous_success_migration": (
"finalize_synchronous_platform_run_success",
"synchronous success profile requires synchronous Run",
"framework_kind = 'legacy'",
"gda.public_dataops_attempt.v1",
"evidence->>'execution_mode' = 'local_inline'",
"content-bound output Artifact was not found",
"independent passed QualityResult was not found",
"input-to-output LineageEvent was not found",
"TO gda_control_gateway",
),
"binding_migration": (
"CREATE TABLE IF NOT EXISTS gda_control.metadata_fabric_binding",
"FOREIGN KEY (tenant_id, execution_plan_artifact_id)",
Expand Down Expand Up @@ -2124,6 +2154,7 @@ def build_gateway_report(
forbidden in role_sql
or forbidden in texts.get("command_migration", "")
or forbidden in texts.get("success_migration", "")
or forbidden in texts.get("synchronous_success_migration", "")
or forbidden in texts.get("binding_migration", "")
or forbidden in texts.get("lineage_migration", "")
):
Expand Down
Loading
Loading