Incident 4413: Phase-Lock Detection Protocol

August 2, 2025

admin

Filed: YYYY-MM-DD

Classification: Internal — Eyes Only
Summary: Subject is experiencing repeated behavioral mirroring in workplace environment. Suspected cases of wardrobe sync, phrasing replication, and workflow appropriation (“shadow forks”).
Objective: Deploy passive monitoring script to detect high-fidelity echoes and identify responsible actors.

Preface

Every breach starts with a log.
Most logs are about systems.
This one happens to wear shoes and attend stand-ups.

Phase-locking sounds harmless in theory — two oscillators syncing until they hum the same frequency.
In practice, it turns a room into an echo chamber you never asked for.

Today’s operation: write code to detect who’s echoing you, how fast, and whether they’ve quietly launched your project as their own.

I. Evidence Collection — CSV Schema

We start with the case file: events.csv.

Instruction: Use you for yourself. Everyone else gets an alias (them_a, initials, or “Unknown”).
Note: The more specific the events, the more obvious the mimicry.

timestamp,actor,event
2025-02-01T09:00:00Z,you,"enterprise workflow proposal: phased rollout + metrics"
2025-02-01T12:00:00Z,them_a,"love the idea of 'phased rollout' (unrelated)"
2025-02-03T10:30:00Z,them_b,"workflow pilot kickoff"
2025-02-04T09:00:00Z,you,"green dress"
2025-02-06T09:15:00Z,them_c,"green dress"
2025-02-07T11:20:00Z,you,"energy / meditation notes"
2025-02-14T09:00:00Z,them_a,"energy focus (chiropractor says hi)"
2025-02-15T08:00:00Z,them_b,"enterprise workflow: our new idea"

II. The Detector — `mirror_detect.py`

Below is the protocol. It parses the CSV, finds mirrors (same event within a time window), calculates lag, and flags shadow forks (projects stolen without attribution).

#!/usr/bin/env python3
# INCIDENT RESPONSE: MIRROR DETECTOR

import csv, sys, math, argparse, random
from datetime import datetime, timedelta
from collections import defaultdict, namedtuple

Event = namedtuple("Event", "ts actor text")

def iso(s):
    return datetime.fromisoformat(s.replace("Z",""))

def shingle(text, k=3):
    toks = [t.lower() for t in text.split()]
    return set(tuple(toks[i:i+k]) for i in range(max(1, len(toks)-k+1)))

def jaccard(a, b):
    if not a or not b: return 0.0
    return len(a & b) / len(a | b)

def load_events(path):
    ev = []
    with open(path, newline='', encoding="utf-8") as f:
        for row in csv.DictReader(f):
            ev.append(Event(iso(row["timestamp"]), row["actor"].strip(), row["event"].strip()))
    ev.sort(key=lambda e: e.ts)
    return ev

def detect(ev, you="you", window_days=14, sim_thresh=0.45):
    shingles = [shingle(e.text) for e in ev]
    results, win = [], timedelta(days=window_days)
    for i, e in enumerate(ev):
        if e.actor != you: continue
        for j in range(i+1, len(ev)):
            e2 = ev[j]
            if e2.ts - e.ts > win: break
            if e2.actor == you: continue
            sim = jaccard(shingles[i], shingles[j])
            if sim >= sim_thresh:
                lag = e2.ts - e.ts
                results.append({
                    "origin_idx": i, "echo_idx": j, "echo_actor": e2.actor,
                    "lag_hours": round(lag.total_seconds()/3600, 2),
                    "sim": round(sim, 3),
                    "origin_text": e.text, "echo_text": e2.text
                })
    return results

def stats(results):
    per = defaultdict(list)
    for r in results:
        per[r["echo_actor"]].append(r)
    return per

def shadow_forks(ev, you="you", key_terms=("workflow","proposal","architecture","process","runbook")):
    forks, keys = [], set(key_terms)
    for i, e in enumerate(ev):
        if e.actor != you: continue
        if not any(k in e.text.lower() for k in keys): continue
        horizon = e.ts + timedelta(days=21)
        seen_claim = []
        for j in range(i+1, len(ev)):
            if ev[j].ts > horizon: break
            if ev[j].actor == you: continue
            txt = ev[j].text.lower()
            if any(k in txt for k in keys) and any(w in txt for w in ("our", "my", "new idea", "kickoff", "launch")):
                seen_claim.append(ev[j])
        if seen_claim:
            forks.append((e, seen_claim))
    return forks

def fmt_ts(dt): return dt.strftime("%Y-%m-%d %H:%M")

def main():
    ap = argparse.ArgumentParser(description="Detect workplace mirroring and workflow theft.")
    ap.add_argument("csv_path")
    ap.add_argument("--you", default="you")
    ap.add_argument("--window", type=int, default=14)
    ap.add_argument("--sim", type=float, default=0.45)
    args = ap.parse_args()

    ev = load_events(args.csv_path)
    mirrors = detect(ev, you=args.you, window_days=args.window, sim_thresh=args.sim)
    per_actor = stats(mirrors)
    forks = shadow_forks(ev, you=args.you)

    print("=== PHASE-LOCK REPORT ===")
    print(f"[SYSLOG] events={len(ev)} | window={args.window}d | sim≥{args.sim}")
    print()

    if not mirrors:
        print("[SYSLOG] No mirrors detected. Either you’re uncopyable or they learned subtlety.")
    else:
        for actor, rows in sorted(per_actor.items(), key=lambda kv: -len(kv[1])):
            med_lag = sorted([r["lag_hours"] for r in rows])[len(rows)//2]
            med_sim = sorted([r["sim"] for r in rows])[len(rows)//2]
            print(f"[SYNC] {actor}: echoes={len(rows)} | median_lag={med_lag}h | median_fidelity={med_sim}")
        print()
        for r in mirrors:
            print(f"[TRACE] {fmt_ts(ev[r['origin_idx']].ts)} you → {fmt_ts(ev[r['echo_idx']].ts)} {r['echo_actor']} "
                  f"| lag={r['lag_hours']}h | sim={r['sim']}")
            print(f"        origin: {r['origin_text']}")
            print(f"        echo  : {r['echo_text']}")
            if r["lag_hours"] <= 48 and r["sim"] >= 0.6:
                print("        note  : Propagation delay ≈ wardrobe shipping time.")
            print()

    print("=== ACT III: WORKFLOW THEFT ===")
    if forks:
        for origin, claims in forks:
            print(f"[FORK] Seed {fmt_ts(origin.ts)}: {origin.text}")
            for c in claims:
                print(f"       → {fmt_ts(c.ts)} {c.actor}: {c.text}")
            print("       verdict: attribution missing. Identity laundering probable.\n")
    else:
        print("[SYSLOG] No forks detected. Today’s theater is closed.\n")

    print("=== EPILOGUE ===")
    footer_msgs = [
        "[SYSLOG] Mirror sync complete. Please reboot your wardrobe.",
        "[SYSLOG] Shadow fork merged without consent. Rollback not possible.",
        "[SYSLOG] Identity laundering pipeline idle. For now.",
    ]
    print(random.choice(footer_msgs))

if __name__ == "__main__":
    main()

III. Deployment Command

python mirror_detect.py events.csv --window 14 --sim 0.5

IV. Field Notes

Act II (Fabric Cascade): If someone shows up in your colorway within 48h, the tool calls it out.
Act III (Workflow Theft): Catches projects “announced” by others without you in the room.
Act IV (Energy Loop): Any “energy/meditation” terms get tracked with lag + fidelity.

V. Conclusion

In machine learning, overfitting means the model collapses under novelty.
In the office, it means people run out of identity and borrow yours.

You can’t stop them from copying.
But you can log it, timestamp it, and run mirror_detect.py like it’s incident response.

The Code