Workspace/Mini projects
Loading progress
All mini projects
MODULE 20 · 5 HOUR BUILD

Proactive incident proposal engine

Build an event-driven monitor that deduplicates failures, groups them into evidence windows, and creates versioned review proposals. The seed prepares proposals and performs no external notification.

Build evidence Record your actual checks, results, and limitations.

Build it in stages

  1. Define event identity, event-time windows, and a late-data policy.
  2. Add deterministic thresholds, minimum sample rules, hysteresis, and cooldown.
  3. Persist admitted events, trigger decisions, and versioned proposals.
  4. Implement standing authorization and revision-bound approval checks at the action boundary.
  5. Inject duplicate deliveries, missing schedules, stale approvals, and uncertain sink outcomes.

Your acceptance criteria

Use these as your project review. Record commands, outputs, and failure cases in your repository.

  • Duplicate delivery does not increase the count or number of proposals.
  • Every proposal lists its window, entity, distinct event IDs, and payload fingerprint.
  • An unapproved or stale proposal causes zero external effects.
  • A replay under the same logical proposal key does not create another proposal.

A working starting point

The seed runs as supplied. Extend it to satisfy the full brief. It is a teaching starting point, not a finished portfolio submission.

main.py
python
import hashlib
import json

EVENTS = [
    {'source': 'ci', 'id': '1', 'entity': 'api', 'time': 10},
    {'source': 'ci', 'id': '1', 'entity': 'api', 'time': 10},
    {'source': 'ci', 'id': '2', 'entity': 'api', 'time': 20},
    {'source': 'ci', 'id': '3', 'entity': 'web', 'time': 25},
]

def fingerprint(value):
    encoded = json.dumps(value, sort_keys=True, separators=(',', ':'))
    return hashlib.sha256(encoded.encode()).hexdigest()

def proposals(events, start, end, threshold, ledger):
    seen, grouped = set(), {}
    for event in events:
        key = (event['source'], event['id'])
        if key in seen:
            continue
        seen.add(key)
        if start <= event['time'] < end:
            grouped.setdefault(event['entity'], []).append('/'.join(key))
    for entity, event_ids in sorted(grouped.items()):
        if len(event_ids) < threshold:
            continue
        key = f'{entity}:{start}:{end}:v1'
        payload = {'entity': entity, 'window': [start, end],
                   'events': sorted(event_ids), 'action': 'draft-investigation'}
        digest = fingerprint(payload)
        if key in ledger and ledger[key]['fingerprint'] != digest:
            raise ValueError('proposal key conflicts with changed evidence')
        ledger.setdefault(key, {'status': 'needs-review', 'payload': payload,
                                'fingerprint': digest})

def main():
    ledger = {}
    proposals(EVENTS, 0, 60, 2, ledger)
    proposals(EVENTS, 0, 60, 2, ledger)
    print(json.dumps(ledger, sort_keys=True))
    print('proposals:', len(ledger))
    print('external effects:', 0)

if __name__ == '__main__':
    main()

Push it further

Add a durable scheduler and outbox, replay a labeled event history, and report detection delay, unnecessary alerts, repeated notifications, and recovery behavior.