GovOPlaN/govoplan-dataflow
2
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Add version-pinned reconciliation decisions

Browse Source
This commit is contained in:
zemion
2026-08-04 11:45:35 +02:00
parent 3991c179a4
commit 314e7c3edf
15 changed files with 724 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_dataflow/backend/executor.py
+212
View File
@@ -979,6 +979,205 @@ def _comparison_fields(value: object) -> tuple[tuple[str, str], ...]:
return tuple(fields)
def _apply_reconciliation_decisions(
records: list[dict[str, Any]],
decisions: list[dict[str, Any]],
config: dict[str, Any],
*,
node_id: str,
) -> tuple[list[dict[str, Any]], tuple[str, ...]]:
columns = {
name: str(config[name])
for name in (
"decision_key_column",
"decision_input_column",
"decision_ref_column",
"action_column",
"actor_column",
"decided_at_column",
"reason_column",
"correction_column",
)
}
allowed_actions = {
str(action)
for action in config.get("allowed_actions", ())
}
decision_index: dict[str, dict[str, Any]] = {}
for row_number, decision in enumerate(decisions, start=1):
key_hash = _required_decision_hash(
decision.get(columns["decision_key_column"]),
row_number=row_number,
label="key hash",
node_id=node_id,
)
_required_decision_hash(
decision.get(columns["decision_input_column"]),
row_number=row_number,
label="input hash",
node_id=node_id,
)
for field_name, label in (
("decision_ref_column", "reference"),
("action_column", "action"),
("actor_column", "actor reference"),
("decided_at_column", "decision time"),
):
_required_decision_text(
decision.get(columns[field_name]),
row_number=row_number,
label=label,
node_id=node_id,
)
action = str(decision[columns["action_column"]]).strip()
if action not in allowed_actions:
raise PipelineExecutionError(
f"Decision row {row_number} uses an action outside the governed action set.",
node_id=node_id,
)
correction = decision.get(columns["correction_column"])
if correction is not None and not isinstance(correction, dict):
raise PipelineExecutionError(
f"Decision row {row_number} correction must be an object or null.",
node_id=node_id,
)
if action == "correct" and not correction:
raise PipelineExecutionError(
f"Decision row {row_number} requires a non-empty correction object.",
node_id=node_id,
)
if key_hash in decision_index:
raise PipelineExecutionError(
"Decision rows contain more than one current decision for a reconciliation key.",
node_id=node_id,
)
decision_index[key_hash] = decision
matched_keys: set[str] = set()
stale_count = 0
output: list[dict[str, Any]] = []
for record in records:
key_hash = str(record.get("_reconciliation_key_hash") or "")
input_hash = str(record.get("_reconciliation_input_hash") or "")
if not _is_sha256(key_hash) or not _is_sha256(input_hash):
raise PipelineExecutionError(
"Decision application requires reconciliation key and input hashes.",
node_id=node_id,
)
decision = decision_index.get(key_hash)
item = dict(record)
if decision is None:
_set_decision_fields(item, state="unreviewed")
output.append(item)
continue
matched_keys.add(key_hash)
decision_input_hash = str(
decision[columns["decision_input_column"]]
).strip()
state = "applied" if decision_input_hash == input_hash else "stale"
if state == "stale":
stale_count += 1
_set_decision_fields(
item,
state=state,
action=str(decision[columns["action_column"]]).strip(),
decision_ref=str(
decision[columns["decision_ref_column"]]
).strip(),
actor_ref=str(decision[columns["actor_column"]]).strip(),
decided_at=str(decision[columns["decided_at_column"]]).strip(),
reason=_optional_decision_text(
decision.get(columns["reason_column"])
),
correction=decision.get(columns["correction_column"]),
)
output.append(item)
unmatched_count = len(decision_index.keys() - matched_keys)
messages = tuple(
message
for count, message in (
(
stale_count,
f"{stale_count} decision(s) are stale because reconciliation inputs changed.",
),
(
unmatched_count,
f"{unmatched_count} decision(s) no longer match a current reconciliation row.",
),
)
if count
)
return output, messages
def _required_decision_hash(
value: object,
*,
row_number: int,
label: str,
node_id: str,
) -> str:
text = str(value or "").strip()
if not _is_sha256(text):
raise PipelineExecutionError(
f"Decision row {row_number} requires a SHA-256 {label}.",
node_id=node_id,
)
return text
def _is_sha256(value: str) -> bool:
if len(value) != 64:
return False
try:
int(value, 16)
except ValueError:
return False
return True
def _required_decision_text(
value: object,
*,
row_number: int,
label: str,
node_id: str,
) -> str:
text = str(value or "").strip()
if not text:
raise PipelineExecutionError(
f"Decision row {row_number} requires a {label}.",
node_id=node_id,
)
return text
def _optional_decision_text(value: object) -> str | None:
text = str(value or "").strip()
return text or None
def _set_decision_fields(
row: dict[str, Any],
*,
state: str,
action: str | None = None,
decision_ref: str | None = None,
actor_ref: str | None = None,
decided_at: str | None = None,
reason: str | None = None,
correction: object | None = None,
) -> None:
row["_decision_state"] = state
row["_decision_action"] = action
row["_decision_ref"] = decision_ref
row["_decision_actor_ref"] = actor_ref
row["_decision_at"] = decided_at
row["_decision_reason"] = reason
row["_decision_correction"] = correction
def _unique_row_index(
rows: list[dict[str, Any]],
columns: list[str],
@@ -1443,6 +1642,18 @@ def _execute_subflow_node(
)
def _execute_reconciliation_decisions(
context: OperatorExecutionContext,
) -> OperatorExecutionResult:
rows, messages = _apply_reconciliation_decisions(
context.outputs[context.inputs_by_port["records"][0]],
context.outputs[context.inputs_by_port["decisions"][0]],
context.node.config,
node_id=context.node.id,
)
return OperatorExecutionResult(rows=rows, messages=messages)
def _register_executors() -> None:
executors = {
"source.inline": _execute_inline_source,
@@ -1546,6 +1757,7 @@ def _register_executors() -> None:
node_id=context.node.id,
)
),
"reconcile.decisions": _execute_reconciliation_decisions,
"subflow": _execute_subflow_node,
"output": lambda context: OperatorExecutionResult(
rows=[dict(row) for row in context.input_rows]
src/govoplan_dataflow/backend/graph.py
+47
View File
@@ -1249,6 +1249,52 @@ def _validate_reconcile(node: GraphNode) -> list[DataflowDiagnostic]:
return diagnostics
def _validate_reconciliation_decisions(
node: GraphNode,
) -> list[DataflowDiagnostic]:
diagnostics: list[DataflowDiagnostic] = []
for field_name in (
"decision_key_column",
"decision_input_column",
"decision_ref_column",
"action_column",
"actor_column",
"decided_at_column",
"reason_column",
"correction_column",
):
if not _non_empty_text(node.config.get(field_name)):
diagnostics.append(
_node_field_error(
node,
"reconcile.decisions.column",
"Every decision mapping column must be named.",
field_name,
)
)
actions = node.config.get("allowed_actions")
if (
not isinstance(actions, list)
or not actions
or len(actions) > 25
or any(
not isinstance(action, str) or not action.strip()
for action in actions
)
or len({action.strip() for action in actions if isinstance(action, str)})
!= len(actions)
):
diagnostics.append(
_node_field_error(
node,
"reconcile.decisions.actions",
"Add between one and 25 unique allowed decision actions.",
"allowed_actions",
)
)
return diagnostics
def _validate_subflow(node: GraphNode) -> list[DataflowDiagnostic]:
diagnostics: list[DataflowDiagnostic] = []
for field_name, code, message in (
@@ -1415,6 +1461,7 @@ def _register_config_validators() -> None:
"limit": _validate_limit,
"quality.rules": _validate_quality,
"reconcile.compare": _validate_reconcile,
"reconcile.decisions": _validate_reconciliation_decisions,
"subflow": _validate_subflow,
"output": _validate_no_config,
}
src/govoplan_dataflow/backend/manifest.py
+1 -1
View File
@@ -200,7 +200,7 @@ DOCUMENTATION = (
"Every graph node declares typed inputs, configuration, output schema, and validation rules. "
"Source nodes pin inline content or governed Datasource references; combine, filter, transform, "
"quality, reconciliation, reusable-subflow, and output nodes remain explicit in the canonical graph. "
"Reconciliation rows expose stable key hashes, explicit before/after values, and input hashes so a later governed human decision can be replayed only while its inputs still match. "
"Reconciliation rows expose stable key hashes, explicit before/after values, and input hashes. A separate decision-table input can annotate exact matches, invalidate changed inputs, and report orphaned decisions without silently rewriting business data. "
"Expressions use the typed Dataflow expression language and never execute arbitrary host or database "
"code. Selecting a node may request a bounded intermediate preview; preview rows are transient, "
"privacy-filtered for the actor, and are not retained as run output. SQL editing compiles into the same "
src/govoplan_dataflow/backend/node_library.py
+79
View File
@@ -530,6 +530,85 @@ _NODE_TYPES = (
},
sql_support="none",
),
NodeTypeDefinition(
type="reconcile.decisions",
category="quality",
label="Apply review decisions",
description="Attach versioned human decisions only while their reconciliation input hash still matches.",
icon="list-checks",
input_ports=(
NodePortDefinition(id="records", label="Reconciliation rows"),
NodePortDefinition(id="decisions", label="Decision rows"),
),
config_fields=(
NodeConfigField(
id="decision_key_column",
label="Decision key hash column",
kind="text",
required=True,
),
NodeConfigField(
id="decision_input_column",
label="Decision input hash column",
kind="text",
required=True,
),
NodeConfigField(
id="decision_ref_column",
label="Decision reference column",
kind="text",
required=True,
),
NodeConfigField(
id="action_column",
label="Action column",
kind="text",
required=True,
),
NodeConfigField(
id="actor_column",
label="Actor reference column",
kind="text",
required=True,
),
NodeConfigField(
id="decided_at_column",
label="Decision time column",
kind="text",
required=True,
),
NodeConfigField(
id="reason_column",
label="Reason column",
kind="text",
required=True,
),
NodeConfigField(
id="correction_column",
label="Correction column",
kind="text",
required=True,
),
NodeConfigField(
id="allowed_actions",
label="Allowed actions",
kind="column_list",
required=True,
),
),
default_config={
"decision_key_column": "key_hash",
"decision_input_column": "input_hash",
"decision_ref_column": "decision_ref",
"action_column": "action",
"actor_column": "actor_ref",
"decided_at_column": "decided_at",
"reason_column": "reason",
"correction_column": "correction",
"allowed_actions": ["accept", "reject", "correct", "defer"],
},
sql_support="none",
),
NodeTypeDefinition(
type="subflow",
category="transform",
src/govoplan_dataflow/backend/schema_validation.py
+62
View File
@@ -640,6 +640,67 @@ def _reconcile(context: SchemaPropagationContext) -> SchemaPropagationResult:
return SchemaPropagationResult(state, tuple(diagnostics))
def _reconciliation_decisions(
context: SchemaPropagationContext,
) -> SchemaPropagationResult:
node = context.node
records = context.port_state("records")
decisions = context.port_state("decisions")
decision_columns = [
str(node.config.get(field_name) or "")
for field_name in (
"decision_key_column",
"decision_input_column",
"decision_ref_column",
"action_column",
"actor_column",
"decided_at_column",
"reason_column",
"correction_column",
)
]
diagnostics = [
*_unknown_columns(
node,
records,
["_reconciliation_key_hash", "_reconciliation_input_hash"],
field_name="records",
),
*_unknown_columns(
node,
decisions,
decision_columns,
field_name="decisions",
),
]
state = SchemaState(
records.columns
| frozenset(
{
"_decision_state",
"_decision_action",
"_decision_ref",
"_decision_actor_ref",
"_decision_at",
"_decision_reason",
"_decision_correction",
}
),
open=records.open,
types={
**records.types,
"_decision_state": "string",
"_decision_action": "string",
"_decision_ref": "string",
"_decision_actor_ref": "string",
"_decision_at": "datetime",
"_decision_reason": "string",
"_decision_correction": "object",
},
)
return SchemaPropagationResult(state, tuple(diagnostics))
def _comparison_columns(value: object) -> tuple[list[str], list[str]]:
left: list[str] = []
right: list[str] = []
@@ -982,6 +1043,7 @@ def register_schema_propagators() -> None:
"limit": _identity,
"quality.rules": _quality,
"reconcile.compare": _reconcile,
"reconcile.decisions": _reconciliation_decisions,
"subflow": _subflow,
"output": _identity,
}
src/govoplan_dataflow/backend/sql_compiler.py
+1
View File
@@ -1953,6 +1953,7 @@ def _register_sql_renderers() -> None:
"limit": _render_limit,
"quality.rules": _render_unsupported,
"reconcile.compare": _render_unsupported,
"reconcile.decisions": _render_unsupported,
"subflow": _render_unsupported,
"output": _render_noop,
}