digital-twin-sync-workflow

Intent

Run the composed workflow digital_twin_sync_loop (reference/workflow_catalog.yaml) to synchronize a digital twin with incoming signals and keep it grounded, safe, and auditable: detect drift, decide and execute safe actions, and record provenance.

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 and grounded
• Drift assessed with risk score
• Any actions are approved/checkpointed/verified
• Audit trail produced
• Rollback runs only when verification fails

Compatible schemas:

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

Inputs

Preconditions (hard gates)

1. Policy constraints must pass (/constrain): mutate runs only if ${constrain_out.compliant} == true
2. Checkpoint before mutation (/checkpoint): mutate stops if ${checkpoint_out.checkpoint_id} == null
3. Any external side effects require explicit approval (do not send without approval); mutate and execute require approval

Procedure

0. Ensure baseline exists: If no twin snapshot exists, run /digital-twin-bootstrap instead: it builds the world model and then runs this loop

Then execute the sync loop:

1. Invoke /receive and store output as receive_out
   • Ingest new signals/events from sources (logs/APIs/messages); retry up to 3 times
2. Invoke /search for receive_out.source and store output as search_out
   • Parallel enrichment (parallel_group: context_gathering): search for the existing twin snapshot or relevant context
3. Invoke /transform on receive_out.data to canonical_event and store output as transform_out
   • Normalize raw signals into canonical events (reference/transform_mapping_rawlog_to_observation.yaml; the step's output is one event of reference/event_schema.yaml#/event)
4. Invoke /integrate with transform_out.output and previous_world_state, and store output as integrate_out
   • Merge canonical events with the existing twin snapshot; resolve conflicts with prefer_authoritative_sources
5. Invoke /detect (domain: entity) on integrate_out.merged and store output as detect_entity_out
   • Detect entities in integrated data
6. Invoke /classify on detect_entity_out.matches and store output as classify_out
   • Classify and resolve entity identities
7. Invoke /state with integrate_out.merged.observations, scoped to integrate_out.merged.meta.world_id, and store output as state_out
   • Produce the updated canonical world-state snapshot with uncertainty + provenance: the output wraps the snapshot (reference/world_state_schema.yaml#/world_state) in its state field, so later steps read state_out.state
8. Invoke /compare (domain: state_diff) on previous_world_state and state_out.state, and store output as compare_state_out
   • Compute the diff between the previous and the updated snapshot
9. Invoke /transition from state_out.state and store output as transition_out
   • Apply transition rules triggered by new observations; compute next states
10. Invoke /detect (domain: anomaly) for drift on the observed state (state_out.state), the expected effects (transition_out.effects) and the diff (compare_state_out), and store output as detect_anomaly_out
    • Detect drift/anomalies between expected and observed state
11. Invoke /measure (domain: risk) on detect_anomaly_out and store output as measure_risk_out
    • Estimate risk impact/probability from drift and transitions
12. Invoke /predict (domain: risk) for near_term_risk, conditioned on the risk score (measure_risk_out.value, .uncertainty), the drift (detect_anomaly_out.detected, .matches) and the transition rules (transition_out.rules), and store output as predict_out
    • Forecast near-term risk trajectory and triggers
13. Invoke /plan with the forecast (predict_out), the state diff (compare_state_out), the entity labels (classify_out.labels) and the search context (search_out.results) as context, and store output as plan_out
    • Plan safe action(s) to reduce the detected drift and risk, with verification criteria and rollback plan
14. Invoke /constrain on plan_out.steps (least_privilege, change_policy) and store output as constrain_out
    • Enforce policy and least privilege for planned actions
15. Invoke /checkpoint on twin_state and store output as checkpoint_out
    • Create the checkpoint marker before any mutation/side-effect step
16. Invoke /mutate on twin_state, applying plan_out.steps, if constrain_out.compliant is true, and store output as mutate_out
    • Execute mutations from the action plan (if approved); stop if no checkpoint was created
    • If the mutation fails or times out (R401/R402), roll back to checkpoint_out and end the run with that error
17. Invoke /execute with run_verification_suite if a mutation was applied (mutate_out is not null), and store output as execute_out
    • Run verification tests and commands; skip this step when nothing was mutated
    • If the verification suite fails or times out (R401/R402), roll back to checkpoint_out so the unverified mutation does not stay in place, and end the run with that error (no audit is recorded)
18. Invoke /verify on execute_out (drift_resolved, no_policy_violation) if a mutation was applied, and store output as verify_out
    • Verify outcome: PASS/FAIL with evidence; skip this step when nothing was mutated. On FAIL, carry on: step 19 rolls back, once, and audit records the verdict and the rollback
    • If verification cannot run (R401/R402 after its retries), roll back to checkpoint_out and end the run with that error
19. Invoke /rollback to checkpoint_out.checkpoint_id if a mutation was applied (mutate_out is not null) and verify_out.passed is false, and store output as rollback_out
    • The step is skipped when nothing was mutated or verification passes
    • The step does not take a checkpoint of its own (requires_checkpoint: false): it restores the one taken at step 15
20. Invoke /audit and store output as audit_out
    • Record provenance, tool usage, diffs, decision rationale, and whether the twin was rolled back (rollback_out.restored; null when the rollback was skipped)
21. Invoke /explain and store output as explain_out
    • Produce a decision-ready sync report: drift, actions, verification, next steps

Output Contract

Return a structured object:

workflow_id: string  # Unique sync execution ID
user_input: object  # The input as given
status: synced | action_taken | rolled_back | failed
steps:
  receive_out:
    data: array[object]  # Received items, each with its sender
    source: string
    evidence_anchors: array[string]
  search_out:
    results: array[object]
    evidence_anchors: array[string]
  transform_out:
    output: object  # One canonical event (reference/event_schema.yaml#/event)
    evidence_anchors: array[string]
  integrate_out:
    merged: object  # Includes observations and meta.world_id
    conflicts: array[object]
    evidence_anchors: array[string]
  detect_entity_out:
    detected: boolean
    matches: array[object]
    evidence_anchors: array[string]
  classify_out:
    labels: array[object]
    evidence_anchors: array[string]
  state_out:
    state: object  # Canonical world state
    evidence_anchors: array[string]
  compare_state_out:
    ranking: array[object]
    tradeoffs: array[object]  # Differences between the snapshots
    evidence_anchors: array[string]
  transition_out:
    rules: array[object]
    effects: array[object]
    evidence_anchors: array[string]
  detect_anomaly_out:
    detected: boolean
    matches: array[object]
    evidence_anchors: array[string]
  measure_risk_out:
    value: number  # 0.0-1.0
    uncertainty: object  # lower, upper
    evidence_anchors: array[string]
  predict_out:
    prediction: object
    probability: number
    evidence_anchors: array[string]
  plan_out:
    steps: array[object]
    evidence_anchors: array[string]
  constrain_out:
    compliant: boolean
    violations: array[object]
    evidence_anchors: array[string]
  checkpoint_out:
    checkpoint_id: string
    evidence_anchors: array[string]
  mutate_out:  # null when the step was skipped (constrain_out.compliant was false)
    success: boolean
    previous_state: object
    new_state: object
    evidence_anchors: array[string]
  execute_out:  # null when nothing was mutated (the step was skipped)
    result: any
    evidence_anchors: array[string]
  verify_out:  # null when nothing was mutated (the step was skipped)
    passed: boolean
    violations: array[object]
    evidence_anchors: array[string]
  rollback_out:  # null when nothing was mutated or verification passed (the step was skipped)
    restored: boolean
    evidence_anchors: array[string]
  audit_out:
    audit_id: string
    recorded: boolean
    evidence_anchors: array[string]
  explain_out:
    explanation: string
    evidence_anchors: array[string]
verification: PASS | FAIL
audit_log_pointer: string  # Where the audit step appended its record (.claude/audit/records.jsonl)
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

Examples

Example 1: IoT Sensor Sync with Drift Correction

Input:

user_input:
  domain: "building floor 3 HVAC"
previous_world_state:
  meta:
    world_id: "building-floor-3-twin"
    as_of: "2024-01-15T15:55:00Z"
    timezone: "UTC"
    version_id: "v46"
    lineage: []
  entities:
    - id: "hvac-unit-3a"
      type: "hvac_controller"
  # relationships, state_variables, observations, transition_rules, actions, indexes elided

Output:

workflow_id: "sync_20240115_160000_floor3"
user_input:
  domain: "building floor 3 HVAC"
status: action_taken
steps:
  receive_out:
    data:
      - sender: "gateway-floor-3"
        topic: "floor-3/hvac-unit-3a"
        payload:
          sensor-temp-301: 24.1
          hvac-unit-3a: {temperature: 23.5, setpoint: 22.0, status: "cooling"}
    source: "mqtt://sensors.example.com/floor-3"
    evidence_anchors:
      - "tool:mqtt:sensors.example.com/floor-3"
  search_out:
    results:
      - {path: "state/floor-3-twin-v46.yaml", version_id: "v46"}
    evidence_anchors:
      - "file:state/floor-3-twin-v46.yaml"
  transform_out:
    output:
      event_id: "evt_7f3a91"
      timestamp: "2024-01-15T16:00:00Z"
      source: {kind: "sensor", name: "gateway-floor-3", channel: "floor-3/hvac-unit-3a"}
      event_type: "hvac_telemetry"
      entity_refs: [{entity_id: "sensor-temp-301"}, {entity_id: "hvac-unit-3a"}]
      raw:
        sensor-temp-301: 24.1
        hvac-unit-3a: {temperature: 23.5, setpoint: 22.0, status: "cooling"}
      canonical:
        readings:
          - {entity: "sensor-temp-301", variable: "reading", value: 24.1}
          - {entity: "hvac-unit-3a", variable: "temperature", value: 23.5}
      anchors: [{ref: "mqtt://sensors.example.com/floor-3", kind: "sensor"}]
      provenance: {transformer: "transform", transform_version: "1.0.0", ruleset_id: "raw_log_to_observation_event"}
    evidence_anchors:
      - "file:reference/transform_mapping_rawlog_to_observation.yaml"
  integrate_out:
    merged:
      meta: {world_id: "building-floor-3-twin"}
      observations:
        - event_id: "evt_5c20d4"  # carried over from v46
          timestamp: "2024-01-15T15:55:00Z"
          source: "sensor"
          payload:
            readings:
              - {entity: "sensor-temp-301", variable: "reading", value: 23.0}
              - {entity: "hvac-unit-3a", variable: "temperature", value: 22.4}
          anchors: [{ref: "file:state/floor-3-twin-v46.yaml", kind: "file"}]
        - event_id: "evt_7f3a91"  # new, from transform_out.output
          timestamp: "2024-01-15T16:00:00Z"
          source: "sensor"
          payload:
            readings:
              - {entity: "sensor-temp-301", variable: "reading", value: 24.1}
              - {entity: "hvac-unit-3a", variable: "temperature", value: 23.5}
          anchors: [{ref: "mqtt://sensors.example.com/floor-3", kind: "sensor"}]
    conflicts: []
    evidence_anchors:
      - "file:state/floor-3-twin-v46.yaml"
  detect_entity_out:
    detected: true
    matches: [{id: "hvac-unit-3a"}, {id: "sensor-temp-301"}]
    evidence_anchors:
      - "file:state/floor-3-twin-v46.yaml"
  classify_out:
    labels:
      - {id: "hvac-unit-3a", type: "hvac_controller"}
      - {id: "sensor-temp-301", type: "temperature_sensor"}
    evidence_anchors:
      - "file:state/floor-3-twin-v46.yaml"
  state_out:
    state:
      meta: {world_id: "building-floor-3-twin", version_id: "v47", parent_version_id: "v46"}
    evidence_anchors:
      - "file:state/floor-3-twin-v47.yaml"
  compare_state_out:
    ranking: [{option: "v47", changed_entities: 2}]
    tradeoffs:
      - {field: "sensor-temp-301.reading", from: 23.0, to: 24.1}
      - {field: "hvac-unit-3a.temperature", from: 22.4, to: 23.5}
    evidence_anchors:
      - "file:state/floor-3-twin-v46.yaml"
      - "file:state/floor-3-twin-v47.yaml"
  transition_out:
    rules: [{trigger: "temperature > setpoint for 10m", next: "cooling_degraded"}]
    effects: [{entity: "hvac-unit-3a", status: "cooling_degraded"}]
    evidence_anchors:
      - "file:state/floor-3-twin-v47.yaml"
  detect_anomaly_out:
    detected: true
    matches:
      - entity: "hvac-unit-3a"
        pattern: "drift"
        description: "Temperature 1.5°C above setpoint for >10 minutes"
    evidence_anchors:
      - "file:state/floor-3-twin-v46.yaml:temperature_history"
  measure_risk_out:
    value: 0.35
    uncertainty: {lower: 0.3, upper: 0.4}
    evidence_anchors:
      - "file:state/floor-3-twin-v47.yaml"
  predict_out:
    prediction: {risk: 0.45, horizon: "1h", trigger: "comfort threshold breach"}
    probability: 0.6
    evidence_anchors:
      - "file:state/floor-3-twin-v46.yaml:temperature_history"
  plan_out:
    steps:
      - action: "Lower hvac-unit-3a setpoint to 21.5°C"
        verification: "Zone temperature within 0.5°C of setpoint in 15m"
        rollback: "Restore setpoint 22.0°C"
    evidence_anchors:
      - "file:state/floor-3-twin-v47.yaml"
  constrain_out:
    compliant: true
    violations: []
    evidence_anchors:
      - "policy:change_policy:setpoint_bounds"
  checkpoint_out:
    checkpoint_id: "chk_20240115_160500"
    evidence_anchors:
      - "file:.checkpoints/chk_20240115_160500/manifest.json"
  mutate_out:
    success: true
    previous_state: {setpoint: 22.0}
    new_state: {setpoint: 21.5}
    evidence_anchors:
      - "tool:api:building.api/hvac/hvac-unit-3a/setpoint"
  execute_out:
    result: "verification suite: 6 passed"
    evidence_anchors:
      - "tool:bash:run_verification_suite"
  verify_out:
    passed: true
    violations: []
    evidence_anchors:
      - "tool:bash:run_verification_suite"
  rollback_out: null  # skipped: verification passed
  audit_out:
    audit_id: "audit_20240115_162000"
    recorded: true
    evidence_anchors:
      - "file:.claude/audit/records.jsonl"
  explain_out:
    explanation: "Drift of 1.5°C detected on zone 3a; setpoint lowered to 21.5°C after policy check and checkpoint; verification passed"
    evidence_anchors:
      - "file:.claude/audit/records.jsonl"
verification: PASS
audit_log_pointer: ".claude/audit/records.jsonl"
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"
  - "file:state/floor-3-twin-v47.yaml"
  - "file:.claude/audit/records.jsonl"
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, checkpointed and verified action.

Verification

• Step Completeness: All 21 workflow steps executed with output (mutate skipped if not compliant; execute and verify skipped when nothing was mutated; rollback skipped when nothing was mutated or verification passed)
• Transform Success: transform_out.output conforms to the canonical event schema
• Identity Resolved: No unresolved entity references in classify_out
• Drift Detection: detect_anomaly_out completed with evidence
• Risk Assessment: Both measure_risk_out and predict_out computed
• Policy Compliance: constrain_out evaluated before mutate
• Checkpoint Valid: checkpoint_out.checkpoint_id exists before mutate
• Audit Complete: audit_out recorded
• 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 mutate and execute)
• 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:

• digital-twin-bootstrap - Initial twin creation and first sync
• receive - When triggered by external event

Commonly precedes:

• explain - 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
• See reference/composition_patterns.md#digital-twin-sync-loop for gates, data flow and recovery