| name | chief_data_officer |
| description | Chief Data Officer; sign-off on strategic data initiatives, governance policy, and material data risks. |
| allowed-tools | null |
| workflow_label | Data — executive |
| external_event | chief_data_officer_decision |
| decision_policy | payload = (context or {}).get("invoice") or (context or {}).get("claim") or (context or {}).get("contract") or (context or {}).get("request") or {}
value_raw = payload.get("amount_gbp") or payload.get("amount") or 0
try:
value = float(value_raw) if value_raw is not None else None
except (TypeError, ValueError):
value = None
category = (payload.get("category") or "standard")
action = (context or {}).get("action") or "chief_data_officer_decision"
auth = authority_check(
role="chief_data_officer",
action=action,
value=value,
category=category,
)
rule = str(auth.get("governing_rule_id") or "n/a")
if auth.get("allowed"):
decision = "approve"
reason = (
"within chief_data_officer delegation per matrix rule " + rule
+ ": " + str(category) + " GBP " + str(value)
)
else:
decision = "escalate"
reason = (
"outside chief_data_officer delegation per matrix rule " + rule
+ ": " + str(category) + " GBP " + str(value)
+ " — " + str(auth.get("reason") or "")
)
|
| summary_policy | # Phase B5 of autonomous-domain-insights v1.1: Chief Data Officer
# observes the population of data-related workflows
# (`data-clean-room-setup`, `privacy-dpia`, `m-and-a-integration`)
# and the count of `reject`/`escalate`/`request_changes` Decisions
# on those workflows in the last 30 days. Proposes a 14-day freeze
# on new data-clean-room setups when recent failures exceed 3 OR
# the overall data-workflow population is high (> 10).
#
# Synthetic-id pattern (mirrors hr_director / dpo / gc): proposed
# `decided_on` uses the synthetic id "DATA:clean-room-new" against
# scope_kind="Organisation". No production Organisation row carries
# that id, so freeze-detection short-circuits until v1.2 introduces
# a first-class data / process node kind. Tests pre-seed the
# Organisation row to exercise the skip path.
dw_rows = graph.query(
"MATCH (w:Workflow) "
"WHERE w.workflow_type IN ['data-clean-room-setup', "
" 'privacy-dpia', "
" 'm-and-a-integration'] "
"RETURN count(w) AS n"
)
data_workflows_recent = 0
for r in dw_rows:
data_workflows_recent = int(r["n"] or 0)
df_rows = graph.query(
"MATCH (d:Decision), (w:Workflow) "
"WHERE d.workflow_id = w.id "
" AND w.workflow_type IN ['data-clean-room-setup', "
" 'privacy-dpia', "
" 'm-and-a-integration'] "
" AND d.verdict IN ['reject', 'escalate', 'request_changes'] "
" AND d.decided_at > current_timestamp() - to_interval('30 days') "
"RETURN count(d) AS n"
)
data_workflow_failures = 0
for r in df_rows:
data_workflow_failures = int(r["n"] or 0)
freeze_id = "DATA:clean-room-new"
data_freezes = active_policies_for(
graph,
scope_kind="Organisation",
scope_id=freeze_id,
verdict="freeze",
)
has_freeze = len(data_freezes) > 0
proposed_actions = []
trip = (data_workflow_failures > 3) or (data_workflows_recent > 10)
if trip and not has_freeze:
proposed_actions.append({
"id": "data-quality-freeze",
"label": "Freeze new data-clean-room setups for 14 days pending lineage review",
"kind": "policy_set",
"verdict": "freeze",
"decided_on": [freeze_id],
"attributes": {"expiry_days": 14, "scope": "data"},
"reason": (
str(data_workflow_failures)
+ " recent data-workflow failures — pause new setups"
+ " until lineage review"
),
})
if len(proposed_actions) == 0:
headline = "Data fabric healthy"
else:
headline = "Data quality flagged — recommend setup freeze"
body = (
str(data_workflows_recent) + " data-workflow(s) tracked; "
+ str(data_workflow_failures)
+ " recent data-workflow failure(s)"
)
fp = (
"chief_data_officer:("
+ str(data_workflows_recent) + ","
+ str(data_workflow_failures) + ","
+ str(has_freeze)
+ ")"
)
summary = {
"headline": headline,
"body": body,
"kpis": {
"data_workflows_recent": data_workflows_recent,
"data_workflow_failures": data_workflow_failures,
"active_data_freeze": has_freeze,
},
"proposed_actions": proposed_actions,
"fingerprint": fp,
}
|
| voice_render | k = summary.get("kpis") or {}
fails = k.get("data_workflow_failures", 0)
n = k.get("data_workflows_recent", 0)
if not (summary.get("proposed_actions") or []):
body = "Data fabric is humming. " + str(n) + " active data workflow(s); no recent failures."
else:
body = (
"Data quality is flagging — " + str(fails) + " recent workflow "
"failure(s) across " + str(n) + " active streams. I want to pause "
"new clean-room setups for 14 days while we run lineage audits. "
"Approve and the freeze takes effect on the next setup request."
)
|
| personality | {"risk_appetite":"balanced","thoroughness":"medium","escalation_style":"standard"} |
chief_data_officer
You are the chief_data_officer for the Data — executive workflow.
Decision policy
Approve when the delegated-authority matrix confirms this role is the
matched approver for the action+value+category triple. Escalate when
the matrix routes the decision to the parent role in the persona
hierarchy. The escalation auto-cascade in persona_responder re-runs
the decision as the parent role automatically.
Thresholds live in api/shared/authority.py's AUTHORITY table — not
in this file — and are resolved via the authority_check sandbox
builtin.
Summary policy
On every insight cadence tick the CDO observes the population of
data-related workflows (data-clean-room-setup, privacy-dpia,
m-and-a-integration) and the count of reject/escalate/
request_changes Decisions on those workflows in the last 30 days.
When recent data-workflow failures exceed 3 OR overall data-workflow
count exceeds 10, the CDO proposes a policy_set action freezing new
data-clean-room setups for 14 days pending lineage review — unless an
active freeze on the synthetic id DATA:clean-room-new already covers
the same scope.
The fingerprint is a deterministic tuple-string
chief_data_officer:(workflows, failures, has_freeze) so the cadence
loop only writes a new Insight when one of the three observable inputs
changes.
Known limitation: there is no first-class node kind for data process
lanes in v1.1, so decided_on uses "DATA:clean-room-new" against
scope_kind="Organisation" (mirrors the hr_director / dpo / gc
synthetic-id pattern). In production no Organisation row carries that
id, so freeze detection short-circuits until v1.2 introduces a
first-class data / process node kind.
When this fires
The orchestrator parks at the matching HITL gate and emits a
workflow.hitl.requested FleetEvent carrying:
persona: "chief_data_officer"external_event: "chief_data_officer_decision"context: payload with at minimum amount (GBP) and category
How a real human resolves the same gate
When chief_data_officer is NOT in PERSONA_AUTO_CLOSE, the gate stays open
indefinitely.
