金大哥 - digital-twin-sync-workflow Skill 详情

Intent

Synchronize a digital twin with real-time signals and keep it grounded, safe, and auditable. This workflow maintains alignment between the real world and its digital representation.

This workflow is designed to model both real and digital systems using the canonical world-state schema:

reference/world_state_schema.yaml

Success criteria:

Twin snapshot updated with latest signals
All drift and anomalies detected and documented
Risk assessed and forecasted with evidence
Actions executed only after checkpoint and approval
Complete audit trail with provenance
Rollback available if verification fails

Compatible schemas:

reference/world_state_schema.yaml
reference/event_schema.yaml
reference/workflow_catalog.yaml

Inputs

| Parameter | Required | Type | Description | |-----------|----------|------|-------------| | sources | Yes | array | Data sources to ingest (APIs, files, streams, sensors) | | world_id | Yes | string | Identifier for the digital twin being synchronized | | constraints | No | object | Policy limits, thresholds, timing constraints | | prior_snapshot | No | object | Previous twin state for delta computation | | sync_mode | No | enum | full | incremental | delta_only (default: incremental) |

Preconditions (hard gates)

Policy constraints must exist (/constrain)
Checkpoint before mutation (/checkpoint)
Any external side effects require explicit approval (do not send without approval)

Procedure

Ensure baseline exists: If no twin snapshot exists, start with /world-model-workflow

Then execute the sync loop:

Invoke /receive → store receive_out
- Ingest signals from configured sources
Invoke /transform to normalize to canonical events → transform_out
- Convert raw signals to event schema format
Invoke /integrate to merge events with prior twin snapshot → integrate_out
- Combine new events with existing state
Invoke /identity-resolution → identity_resolution_out
- Resolve entity references across sources
Invoke /world-state producing canonical snapshot → world_state_out
- Generate updated twin representation
Invoke /state-transition apply rules → state_transition_out
- Apply business rules and state machine logic
Invoke /detect-anomaly drift detection → detect_anomaly_out
- Identify deviations from expected behavior
Invoke /estimate-risk risk estimate → estimate_risk_out
- Assess current risk based on anomalies
Invoke /forecast-risk risk forecast → forecast_risk_out
- Project future risk trajectory
Invoke /plan remediation plan + verification criteria + rollback plan → plan_out
- Create action plan if intervention needed
Invoke /constrain enforce policy constraints → constrain_out
- Validate plan against policy limits
Invoke /checkpoint create mutation gate marker → checkpoint_out
- Establish restore point before action
Invoke /act-plan execute if safe/approved → act_plan_out
- Execute remediation actions
Invoke /verify PASS/FAIL → verify_out
- Confirm actions achieved intended outcome
Invoke /audit provenance + tool log → audit_out
- Record complete audit trail
If FAIL or side effects → /rollback → rollback_out
- Restore previous state if needed
Invoke /summarize decision-ready report → summarize_out
- Generate executive summary

Output Contract

Return a structured object:

workflow_id: string  # Unique sync execution ID
world_id: string  # Digital twin identifier
sync_timestamp: string  # ISO timestamp of sync
status: synced | drift_detected | action_taken | rolled_back | failed
twin_snapshot:
  version: string
  state: object  # Canonical world state
  hash: string  # Integrity hash
  evidence_anchors: array[string]
drift_report:
  anomalies_detected: integer
  severity: low | medium | high | critical
  triggers: array[string]
  evidence_anchors: array[string]
risk_assessment:
  current_risk: number  # 0.0-1.0
  forecasted_risk: number  # 0.0-1.0
  risk_factors: array[string]
  evidence_anchors: array[string]
actions:
  executed: boolean
  plan_summary: string
  changes: array[string]
  safety_gates_passed: boolean
  evidence_anchors: array[string]
verification:
  result: PASS | FAIL | SKIPPED
  criteria_met: array[string]
  evidence_anchors: array[string]
audit:
  log_path: string
  provenance_chain: array[string]
  evidence_anchors: array[string]
rollback:
  available: boolean
  executed: boolean
  restore_point: string | null
  command: string | null
next_sync:
  recommended_interval: string  # e.g., "5m", "1h"
  triggers: array[string]  # Conditions for immediate resync
confidence: number  # 0.0-1.0
evidence_anchors: array[string]
assumptions: array[string]

Field Definitions

| Field | Type | Description | |-------|------|-------------| | workflow_id | string | Unique identifier for this sync execution | | world_id | string | Digital twin being synchronized | | twin_snapshot | object | Updated canonical world state with integrity hash | | drift_report | object | Detected anomalies and their severity | | risk_assessment | object | Current and forecasted risk levels | | actions | object | What remediation was taken (if any) | | verification | object | Whether actions achieved intended outcome | | audit | object | Complete provenance and audit trail | | rollback | object | Rollback availability and status | | next_sync | object | Recommended timing for next synchronization | | confidence | number | 0.0-1.0 based on evidence quality | | evidence_anchors | array | All evidence references collected | | assumptions | array | Explicit assumptions made during sync |

Examples

Example 1: IoT Sensor Sync with Anomaly Detection

Input:

sources:
  - type: mqtt
    endpoint: "mqtt://sensors.example.com/floor-3"
  - type: api
    endpoint: "https://building.api/hvac/status"
world_id: "building-floor-3-twin"
constraints:
  max_drift_threshold: 0.15
  require_approval_above_risk: 0.7
sync_mode: incremental

Output:

workflow_id: "sync_20240115_160000_floor3"
world_id: "building-floor-3-twin"
sync_timestamp: "2024-01-15T16:00:00Z"
status: drift_detected
twin_snapshot:
  version: "v47"
  state:
    entities:
      - id: "hvac-unit-3a"
        type: "hvac_controller"
        temperature: 23.5
        setpoint: 22.0
        status: "cooling"
      - id: "sensor-temp-301"
        type: "temperature_sensor"
        reading: 24.1
        last_updated: "2024-01-15T15:59:45Z"
    relationships:
      - subject: "hvac-unit-3a"
        predicate: "controls"
        object: "zone-3a"
  hash: "sha256:abc123def456..."
  evidence_anchors:
    - "tool:mqtt:sensors.example.com/floor-3"
    - "tool:api:building.api/hvac/status"
drift_report:
  anomalies_detected: 1
  severity: medium
  triggers:
    - "Temperature 1.5°C above setpoint for >10 minutes"
  evidence_anchors:
    - "file:state/floor-3-twin-v46.yaml:temperature_history"
    - "tool:detect-anomaly:threshold_breach"
risk_assessment:
  current_risk: 0.35
  forecasted_risk: 0.45
  risk_factors:
    - "HVAC may be undersized for current load"
    - "Trending toward comfort threshold breach"
  evidence_anchors:
    - "tool:estimate-risk:hvac_capacity"
    - "tool:forecast-risk:temperature_trend"
actions:
  executed: false
  plan_summary: "Monitor for 15 more minutes before intervention"
  changes: []
  safety_gates_passed: true
  evidence_anchors:
    - "tool:plan:remediation_decision"
verification:
  result: SKIPPED
  criteria_met: []
  evidence_anchors: []
audit:
  log_path: ".claude/audit/sync_20240115_160000_floor3.log"
  provenance_chain:
    - "mqtt://sensors.example.com → receive"
    - "receive → transform"
    - "transform → integrate"
    - "integrate → world-state"
  evidence_anchors:
    - "file:.claude/audit/sync_20240115_160000_floor3.log"
rollback:
  available: false
  executed: false
  restore_point: null
  command: null
next_sync:
  recommended_interval: "5m"
  triggers:
    - "Temperature exceeds 25°C"
    - "HVAC status changes"
confidence: 0.88
evidence_anchors:
  - "tool:mqtt:sensors.example.com/floor-3"
  - "tool:api:building.api/hvac/status"
  - "tool:detect-anomaly:threshold_breach"
  - "file:.claude/audit/sync_20240115_160000_floor3.log"
assumptions:
  - "Sensor readings are accurate within ±0.1°C"
  - "MQTT connection is reliable"
  - "Building API returns real-time status"

Evidence pattern: Multi-source signal ingestion, anomaly detection against historical baseline, risk forecasting with trend analysis.

Verification

[ ] Source Ingestion: All configured sources successfully polled
[ ] Transform Success: Events conform to canonical schema
[ ] Identity Resolved: No unresolved entity references
[ ] Drift Detection: Anomaly check completed with evidence
[ ] Risk Assessment: Both current and forecast risk computed
[ ] Policy Compliance: Constraints checked before any action
[ ] Checkpoint Valid: Restore point exists if actions taken
[ ] Audit Complete: Provenance chain documented
[ ] Next Sync Scheduled: Recommended interval provided

Verification tools: Web (for API checks), Bash (for MQTT), Read (for state files)

Safety Constraints

mutation: true
requires_checkpoint: true
requires_approval: true (for actions above risk threshold)
risk: high

Capability-specific rules:

STOP on low confidence from any perception step
NEVER execute mutation without checkpoint
NEVER emit external side effects without explicit approval
Validate all source data before integration
Preserve previous snapshot for rollback
Rate-limit sync frequency to prevent thrashing

Composition Patterns

Commonly follows:

world-model-workflow - Initial twin creation before first sync
receive - When triggered by external event

Commonly precedes:

summarize - Create executive report after sync
send - Notify stakeholders of drift or actions
Self (recursive) - Next sync iteration

Anti-patterns:

Never sync without prior world model established
Never skip anomaly detection to proceed directly to action
Never execute actions without policy constraint check
Never delete audit logs before retention period

Workflow references:

See reference/workflow_catalog.yaml#digital-twin-sync-loop for step definitions
See reference/world_state_schema.yaml for canonical state format
See reference/composition_patterns.md#checkpoint-act-verify-rollback for CAVR pattern