Temporal detection

A versioned rule fixes the order, group, window, and state budget

The rule names each stage by event kind, the stage it follows, and how much event time may separate them. This one groups by session and looks for egress within ten seconds of a credential access; the timing and severity are choices for one workload.

{
  "rule_id": "credential-then-egress",
  "policy_version": "pricing-v1",
  "group_by": "session_id",
  "max_groups": 128,
  "max_partial_matches_per_group": 16,
  "allow_event_reuse": false,
  "stages": [
    {
      "name": "credential",
      "event_kind": "credential_access",
      "minimum_severity": "low"
    },
    {
      "name": "outbound",
      "event_kind": "egress_attempt",
      "minimum_severity": "low",
      "after": "credential",
      "within_ms": 10000
    }
  ]
}

TemporalRule::parse_json reads the document under explicit RuleLimits. It refuses unknown fields and malformed stage relationships, and checks every bound before the rule is usable. The rule's canonical version hash is computed over the parsed document and becomes part of every finding and every retained partial match.

The first stage starts the sequence, so it carries no after or within_ms. Each later stage names a prior stage and a positive, bounded window. The host installs the parsed rule into its verified-event pipeline and its durable correlation store.

Each partial match occupies retained state, so the operator also sets a state ceiling at parse time. Saved as credential-then-egress.json in a Chio host that depends on chio-quarantine, the document loads under limits of four stages, a sixty-second maximum stage window, 128 groups, and 4,096 estimated state entries:

use chio_quarantine::rules::{RuleLimits, TemporalRule};

fn load_credential_egress_rule() -> Result<TemporalRule, Box<dyn std::error::Error>> {
    let bytes = std::fs::read("credential-then-egress.json")?;
    let limits = RuleLimits::new(4, 60_000, 128, 4_096)?;
    Ok(TemporalRule::parse_json(&bytes, &limits)?)
}

The parser multiplies groups, partial matches per group, and stages with checked arithmetic: 128 × 16 × 2 = 4,096 entries, exactly the limit. Raising max_partial_matches_per_group to seventeen demands 4,352 and fails before any event is ingested; a third stage without a smaller bound elsewhere exceeds the budget too. Raise a limit only after deciding what extra retained work the deployment can carry.

Exercise the two boundaries independently. Validate the two-stage document, raise the partial-match count to seventeen under the same limits, then set the second stage's after to its own name or to an absent stage. The parser refuses each variant, so state budgeting and sequence structure are tested apart.

The window is inclusive and measured in event time

Take every event as verified, in one session, and above the minimum severity. The figure's five histories then resolve by one comparison: the egress's event time minus the credential read's must fall between zero and 10,000 milliseconds.

• An egress 7 seconds after the credential read matches.
• An egress at exactly 10 seconds matches too, because the boundary is included. Eleven seconds is outside the window.
• An egress that precedes the credential read establishes no order, and a credential read in session A never pairs with an egress in session B.

Event time is when the producer observed the action; ingest time is when the correlator received it. Watermarks and the configured lateness bound decide which partial histories can still accept a stage, so an event arriving past that bound is kept out of any sequence the correlator has already settled.

A late arrival is a separate outcome from an event outside the rule window, so check the detector's health and the lateness policy before reading meaning into a missing finding. Clock assumptions belong to the producer and its verification contract; sorting caller-supplied timestamps produces no trustworthy temporal evidence.

The group key decides which events can share a sequence

A rule groups by agent, capability, lineage seed, session, or subject. The tenant comes from the trusted partition every event carries, and rules group within it.

Choose the narrowest identity that captures the behavior. Session grouping isolates one interaction and can miss a relationship that crosses delegated work. Lineage grouping follows delegated work, so events from a worker and the coordinator that delegated to it can share one sequence. A wider identity raises false matches and state demand together, so test legitimate activity at the same boundary.

With allow_event_reuse false, an event joins any candidate at most once. Otherwise one credential read could produce a finding for every egress after it and make a sequence look more frequent than it was. Test that consumption rule alongside the elapsed-time comparison.

A finding names the rule version and the evidence it matched

A complete match produces a CorrelatedFinding, and a response plan binds that exact record:

pub struct CorrelatedFinding {
    pub finding_id: RecordId,
    pub tenant_id: TenantId,
    pub rule_id: RuleId,
    pub rule_version_hash: Digest32,
    pub policy_version: RecordId,
    pub group_key_hash: Digest32,
    pub ordered_event_ids: FindingEventIds,
    pub ordered_evidence_digests: FindingEvidenceDigests,
    pub ordered_source_receipt_ids: FindingSourceReceiptIds,
    pub first_event_time_unix_ms: u64,
    pub last_event_time_unix_ms: u64,
    pub lineage_seed: LineageId,
}

The finding fixes the rule and its version hash, the policy version, the group commitment, the ordered event IDs, the ordered evidence digests, the source receipts, the first and last event times, and the lineage seed.

A partial match belongs to the rule version that read its first events. Changing the window, the grouping key, or the reuse policy yields a new version and a new interpretation, and an old partial sequence stays bound to the version that evaluated it. Before a stream justifies automatic response, review its group and partial-match demand during ordinary bursts as well as attacks.

Seven and ten seconds match across a restart; eleven does not

The comparison authenticates each fixture event, writes the first stage to a SQLite store, closes it, reopens the same database, and evaluates the second stage. Seven, ten, and eleven seconds each get their own database, so event reuse stays out of the boundary test.

Add the module as temporal_cases to a host that uses chio-quarantine and chio-store-sqlite, with the signed-observation fixture as security_event_cases. Call it with a fresh, private, disposable directory, never a serving store. Lateness is zero for this ordered fixture; a deployment sets its own clock-skew and arrival contract.

temporal_cases::compare_temporal_windows(&directory)?;

MATCH 7 s after reopen: original ordered events and evidence retained
MATCH inclusive 10 s boundary; NO MATCH at 11 s in an independent history

You should see both lines. The assertions behind them check both ordered event IDs and both evidence digests, and the retained outcome reads back identically to the returned one. The eleven-second history stays healthy with no finding, because a sequence outside the reviewed window is a result, and the run grants no permission to apply a response.

use crate::security_event_cases::{fixture_verifier, signed_observation, OBSERVATION_TIME};
use chio_core_types::Keypair;
use chio_quarantine::{
    CorrelationOutcome, CorrelationPolicy, RuleLimits, TemporalCorrelator, TemporalRule,
};
use chio_security_types::ports::SecurityEventVerifierPort;
use chio_security_types::SecurityEventKind;
use chio_store_sqlite::security_state::SqliteSecurityStateStore;
use std::{path::Path, sync::Arc};

pub fn credential_egress_rule() -> Result<TemporalRule, Box<dyn std::error::Error>> {
    Ok(TemporalRule::parse_json(
        br#"{
      "rule_id":"credential-then-egress","policy_version":"pricing-v1",
      "group_by":"session_id","max_groups":128,
      "max_partial_matches_per_group":16,"allow_event_reuse":false,
      "stages":[
        {"name":"credential","event_kind":"credential_access","minimum_severity":"low"},
        {"name":"outbound","event_kind":"egress_attempt","minimum_severity":"low",
         "after":"credential","within_ms":10000}]
    }"#,
        &RuleLimits::new(4, 60_000, 128, 4_096)?,
    )?)
}

/// A fresh disposable database per independent candidate history.
/// The same verified first stage survives an actual SQLite close/reopen.
fn candidate(
    path: &Path,
    elapsed_ms: u64,
) -> Result<CorrelationOutcome, Box<dyn std::error::Error>> {
    if path.exists() {
        return Err("candidate database must be fresh".into());
    }
    let producer = Keypair::generate();
    let verifier = fixture_verifier(&producer)?;
    let rule = credential_egress_rule()?;
    let policy = CorrelationPolicy::new(0, 4_096, 3, false)?;
    let first = verifier.verify(&signed_observation(
        &producer,
        "credential-1",
        "review-1",
        SecurityEventKind::CredentialAccess,
        OBSERVATION_TIME,
    )?)?;
    let second = verifier.verify(&signed_observation(
        &producer,
        "egress-1",
        "review-1",
        SecurityEventKind::EgressAttempt,
        OBSERVATION_TIME + elapsed_ms,
    )?)?;
    {
        let detector =
            TemporalCorrelator::new(Arc::new(SqliteSecurityStateStore::open(path)?), policy);
        let partial = detector.ingest(&rule, &first);
        assert!(partial.findings.is_empty());
        assert!(partial.detector_health.is_empty());
        assert!(!partial.automatic_response_suppressed);
    }
    let detector = TemporalCorrelator::new(Arc::new(SqliteSecurityStateStore::open(path)?), policy);
    let outcome = detector.ingest(&rule, &second);
    assert!(outcome.detector_health.is_empty());
    assert!(!outcome.automatic_response_suppressed);
    let retained = detector
        .load_durable_outcome(&rule, &second)?
        .ok_or("no retained outcome")?;
    assert_eq!(
        serde_json::to_vec(&retained)?,
        serde_json::to_vec(&outcome)?
    );
    if let Some(finding) = outcome.findings.first() {
        assert_eq!(
            finding.ordered_event_ids.as_slice(),
            &[first.event_id, second.event_id]
        );
        assert_eq!(
            finding.last_event_time_unix_ms - finding.first_event_time_unix_ms,
            elapsed_ms
        );
        assert_eq!(
            finding.ordered_evidence_digests.as_slice(),
            &[first.evidence_hash, second.evidence_hash]
        );
    }
    Ok(outcome)
}

pub fn compare_temporal_windows(directory: &Path) -> Result<(), Box<dyn std::error::Error>> {
    let seven = candidate(&directory.join("seven.sqlite"), 7_000)?;
    let ten = candidate(&directory.join("ten.sqlite"), 10_000)?;
    let eleven = candidate(&directory.join("eleven.sqlite"), 11_000)?;
    assert_eq!(seven.findings.len(), 1);
    assert_eq!(ten.findings.len(), 1);
    assert!(eleven.findings.is_empty());
    println!("MATCH 7 s after reopen: original ordered events and evidence retained");
    println!("MATCH inclusive 10 s boundary; NO MATCH at 11 s in an independent history");
    Ok(())
}

A missing finding is read together with detector health

A detector that loses events, overflows its state, or cannot reach its store reports that as typed health evidence and suppresses automatic response for the affected scope. Read each correlation result against that evidence before acting on it.

During recovery the production consumer re-verifies retained envelopes, and it publishes signed correlation evidence before acknowledging response work. Keep regression cases beside the positive sequence: reversed order, the time boundary, duplicates, group separation, capacity, and restart, including a restart with exhausted or unavailable correlation storage whose health state must stay visible.

Read more in Detector health, which names each failure kind and how it surfaces.

Next: turn a finding into a permitted action