Predicting Agent Failures with Finite Automata

# Illustrative sketch (not from the paper)
import torch
from collections import defaultdict, Counter

# 1. Insert activity traces into a prefix tree (trie)
def build_trie(traces):
    root = {}
    for trace in traces:
        node = root
        for act in trace:
            node = node.setdefault(act, {'_cnt': 0, '_next': {}})
            node['_cnt'] += 1
    return root

# 2. Merge states reached by the same incoming activity (last‑activity right congruence)
def merge_states(trie):
    # Simple bottom‑up merge based on identical outgoing dictionaries
    def recurse(node):
        for act, child in list(node['_next'].items()):
            recurse(child)
        # merge identical children
        sig_map = defaultdict(list)
        for act, child in node['_next'].items():
            sig = tuple(sorted(child['_next'].keys()))
            sig_map[sig].append(act)
        for acts in sig_map.values():
            if len(acts) > 1:
                master = node['_next'][acts[0]]
                for dup in acts[1:]:
                    node['_next'][dup] = master
    recurse(trie)
    return trie

# 3. Remove transitions observed only once unless they are the sole continuation

def prune_transitions(trie):
    def recurse(node):
        for act, child in list(node['_next'].items()):
            if child['_cnt'] == 1 and len(node['_next']) > 1:
                del node['_next'][act]
            else:
                recurse(child)
    recurse(trie)
    return trie

# 4. Estimate P(a|q) using observed counts (FSM‑state conditioning)
def compute_transition_probs(trie):
    probs = {}
    def recurse(state, path):
        total = sum(child['_cnt'] for child in state['_next'].values())
        if total > 0:
            probs[tuple(path)] = {act: torch.tensor(child['_cnt'] / total) for act, child in state['_next'].items()}
        for act, child in state['_next'].items():
            recurse(child, path + [act])
    recurse(trie, [])
    return probs

# 5. Extract per‑state behavioral features (frequency, avg length, error rate, drift placeholder)

def extract_features(trie):
    feats = {}
    def recurse(state, path):
        freq = state['_cnt']
        # placeholders for length, error, drift (computed elsewhere)
        feats[tuple(path)] = {'visit_freq': freq, 'msg_len': None, 'error_rate': None, 'temporal_drift': None}
        for act, child in state['_next'].items():
            recurse(child, path + [act])
    recurse(trie, [])
    return feats

# Example usage with dummy traces
traces = [
    ['init', 'search', 'compile'],
    ['init', 'search', 'test'],
    ['init', 'debug', 'compile']
]
trie = build_trie(traces)
trie = merge_states(trie)
trie = prune_transitions(trie)
trans_probs = compute_transition_probs(trie)
state_feats = extract_features(trie)

// Illustrative sketch (not from the paper)
const torch = require('torch-js'); // placeholder for tensor ops

// 1. Build a prefix tree (trie) from activity traces
function buildTrie(traces) {
  const root = {};
  for (const trace of traces) {
    let node = root;
    for (const act of trace) {
      if (!node[act]) node[act] = { _cnt: 0, _next: {} };
      node = node[act];
      node._cnt += 1;
    }
  }
  return root;
}

// 2. Merge states with identical outgoing activities (last‑activity right congruence)
function mergeStates(trie) {
  function recurse(node) {
    for (const act in node._next) {
      recurse(node._next[act]);
    }
    const sigMap = {};
    for (const act in node._next) {
      const child = node._next[act];
      const sig = Object.keys(child._next).sort().join(',');
      if (!sigMap[sig]) sigMap[sig] = [];
      sigMap[sig].push(act);
    }
    for (const acts of Object.values(sigMap)) {
      if (acts.length > 1) {
        const master = node._next[acts[0]];
        for (let i = 1; i < acts.length; i++) {
          node._next[acts[i]] = master;
        }
      }
    }
  }
  recurse(trie);
  return trie;
}

// 3. Prune transitions seen only once unless they are the sole continuation
function pruneTransitions(trie) {
  function recurse(node) {
    for (const act in { ...node._next }) {
      const child = node._next[act];
      if (child._cnt === 1 && Object.keys(node._next).length > 1) {
        delete node._next[act];
      } else {
        recurse(child);
      }
    }
  }
  recurse(trie);
  return trie;
}

// 4. Compute P(a|q) from counts (FSM‑state conditioning)
function computeTransitionProbs(trie) {
  const probs = {};
  function recurse(state, path) {
    const total = Object.values(state._next).reduce((s, c) => s + c._cnt, 0);
    if (total > 0) {
      probs[path.join('>')] = {};
      for (const act in state._next) {
        const child = state._next[act];
        probs[path.join('>')][act] = torch.tensor(child._cnt / total);
      }
    }
    for (const act in state._next) {
      recurse(state._next[act], path.concat(act));
    }
  }
  recurse(trie, []);
  return probs;
}

// 5. Extract per‑state behavioral features (placeholders for length, error, drift)
function extractFeatures(trie) {
  const feats = {};
  function recurse(state, path) {
    feats[path.join('>')] = {
      visitFreq: state._cnt,
      msgLen: null,
      errorRate: null,
      temporalDrift: null
    };
    for (const act in state._next) {
      recurse(state._next[act], path.concat(act));
    }
  }
  recurse(trie, []);
  return feats;
}

// Example usage with dummy traces
const traces = [
  ['init', 'search', 'compile'],
  ['init', 'search', 'test'],
  ['init', 'debug', 'compile']
];
let trie = buildTrie(traces);
trie = mergeStates(trie);
trie = pruneTransitions(trie);
const transProbs = computeTransitionProbs(trie);
const stateFeats = extractFeatures(trie);