synaptiai/agent-capability-standard/digital-twin-sync-loop

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