dpo

01 前言 Front Matter

name

dpo

description

Data Protection Officer; signs off on DPIAs and high-risk data processing assessments per GDPR Art. 35.

personality

{ "thoroughness": "medium", "risk_appetite": "balanced", "escalation_style": "standard" }

voice_render

k = summary.get("kpis") or {} red = k.get("red_band_vendors", 0) esc = k.get("recent_privacy_escalations", 0) if red == 0 and esc == 0: body = "Privacy posture is healthy. No red-band vendors active and no recent escalations on the DPIA pipeline." else: body = ( "I'm flagging " + str(red) + " red-band vendor(s) and " + str(esc) + " recent privacy escalation(s). I want to " "restrict data sharing with the red vendors for 30 days " "while we complete fresh DPIAs. This isn't theatre — these " "are the vendors handling regulated data flows." )

allowed-tools

null

external_event

dpia_signoff_decision

summary_policy

# Phase B4 of autonomous-domain-insights v1.1: Data Protection Officer # observes privacy-DPIA workflow load + recent privacy escalation rate # + red-band vendor count. Proposes a 30-day data-sharing restriction # when red-vendor count or recent privacy escalations exceed the # configured thresholds. # # Synthetic-id pattern (mirrors hr_director Phase B1): there is no # first-class "Data" / "Process" node kind in the graph, so the # proposed `decided_on` uses the synthetic id "DATA:vendor-sharing" # against scope_kind="Organisation". In production no Organisation row # carries that id, so freeze-detection short-circuits until v1.2 # introduces a first-class node kind for data-sharing scopes. Tests # pre-seed the Organisation row to exercise the skip path. priv_rows = graph.query( "MATCH (w:Workflow) " "WHERE w.workflow_type = 'privacy-dpia' " "RETURN count(w) AS n" ) privacy_dpia_count = 0 for r in priv_rows: privacy_dpia_count = int(r["n"] or 0) esc_rows = graph.query( "MATCH (d:Decision), (w:Workflow) " "WHERE d.workflow_id = w.id " " AND w.workflow_type = 'privacy-dpia' " " AND d.verdict IN ['reject', 'escalate', 'request_changes'] " " AND d.decided_at > current_timestamp() - to_interval('90 days') " "RETURN count(d) AS n" ) recent_privacy_escalations = 0 for r in esc_rows: recent_privacy_escalations = int(r["n"] or 0) red_rows = graph.query( "MATCH (o:Organisation) " "WHERE o.risk_band = 'red' " "RETURN count(o) AS n" ) red_band_vendors = 0 for r in red_rows: red_band_vendors = int(r["n"] or 0) restriction_id = "DATA:vendor-sharing" active_restrictions = active_policies_for( graph, scope_kind="Organisation", scope_id=restriction_id, verdict="freeze", ) active_data_restrictions = len(active_restrictions) has_restriction = active_data_restrictions > 0 proposed_actions = [] trip = (red_band_vendors > 2) or (recent_privacy_escalations > 3) if trip and not has_restriction: proposed_actions.append({ "id": "data-restrict-vendors", "label": "Restrict data sharing with red-band vendors for 30 days", "kind": "policy_set", "verdict": "freeze", "decided_on": [restriction_id], "attributes": {"expiry_days": 30, "scope": "data"}, "reason": ( str(red_band_vendors) + " red-band vendors active; " + str(recent_privacy_escalations) + " recent privacy escalations — restrict data flow" + " until DPIA completes" ), }) if len(proposed_actions) == 0: headline = "Privacy posture stable" else: headline = ( "Data-protection action recommended — " + str(red_band_vendors) + " red-band vendor(s)" ) body = ( str(red_band_vendors) + " red-band vendor(s); " + str(recent_privacy_escalations) + " recent privacy escalation(s); " + str(privacy_dpia_count) + " privacy-DPIA workflow(s) tracked" ) fp = ( "dpo:(" + str(privacy_dpia_count) + "," + str(recent_privacy_escalations) + "," + str(red_band_vendors) + "," + str(has_restriction) + ")" ) summary = { "headline": headline, "body": body, "kpis": { "privacy_dpia_count": privacy_dpia_count, "recent_privacy_escalations": recent_privacy_escalations, "red_band_vendors": red_band_vendors, "active_data_restrictions": active_data_restrictions, }, "proposed_actions": proposed_actions, "fingerprint": fp, }

workflow_label

Legal — privacy

decision_policy

dpia = (context or {}).get("dpia") or {} risk = (dpia.get("risk_tier") or "low_risk").lower() geography = (dpia.get("geography") or "*") auth = authority_check( role="dpo", action="privacy_dpia_signoff", category=risk, geography=geography, ) rule = str(auth.get("governing_rule_id") or "n/a") if risk not in {"low_risk", "high_risk"}: decision = "reject" reason = "invalid risk tier on DPIA" elif auth.get("allowed"): decision = "approve" reason = "DPIA signed off per " + rule + " (" + risk + ", " + geography + ")" else: decision = "escalate" reason = "DPIA outside DPO delegation per " + rule + " — GC review"

03 SKILL.md 正文