diff --git a/src/flow.py b/src/flow.py index e46c2a3..adeb087 100644 --- a/src/flow.py +++ b/src/flow.py @@ -826,42 +826,6 @@ def _embed_persona(args): except Exception as e: logger.error(f"[genesis] Persona embed failed for {name}: {e}") - sim_start = datetime.strptime(CONFIG["simulation"]["start_date"], "%Y-%m-%d") - for gap in CONFIG.get("knowledge_gaps", []): - name = gap["name"] - left_date = gap["left"] # "2024-06" - left_dt = datetime.strptime(left_date, "%Y-%m") - departure_day = -(sim_start - left_dt).days - - self._mem.log_event( - SimEvent( - type="employee_departed", - day=departure_day, - date=f"{left_date}-01", - timestamp=f"{left_date}-01T09:00:00", - actors=[name], - artifact_ids={}, - facts={ - "name": name, - "role": gap.get("role", ""), - "knowledge_domains": gap.get("knew_about", []), - "documented_pct": gap.get("documented_pct", 0.0), - "reason": "voluntary", - "scheduled": True, - }, - summary=( - f"{name} ({gap.get('role', 'unknown role')}) departed " - f"Day {departure_day} [voluntary]. " - f"Gaps: {', '.join(gap.get('knew_about', []))}. " - f"~{int(gap.get('documented_pct', 0) * 100)}% documented." - ), - tags=["employee_departed", "lifecycle", "genesis"], - ) - ) - logger.info( - f"[genesis] Logged pre-sim departure: {name} (Day {departure_day})" - ) - self._confluence.write_genesis_batches_parallel( [ { @@ -1790,33 +1754,7 @@ def _handle_incident(self): on_call, ) - involves_gap = any( - k.lower() in root_cause.lower() - for emp in DEPARTED_EMPLOYEES.values() - for k in emp["knew_about"] - ) - - gap_areas: List[str] = [] - gap_context_str: str = "" - if involves_gap: - departed_details: List[str] = [] - for emp_name, emp in DEPARTED_EMPLOYEES.items(): - hits = [k for k in emp["knew_about"] if k.lower() in root_cause.lower()] - if hits: - gap_areas.extend(hits) - departed_details.append( - f"{emp_name} (ex-{emp['role']}, left {emp['left']}, " - f"{int(emp['documented_pct'] * 100)}% documented) " - f"owned: {', '.join(hits)}" - ) - if departed_details: - gap_context_str = ( - f"KNOWLEDGE GAP FLAG: This incident touches underdocumented systems. " - f"{' | '.join(departed_details)}. " - f"Resolution may be blocked pending knowledge recovery." - ) - - self._lifecycle.scan_for_knowledge_gaps( + detected_gaps = self._lifecycle.scan_for_knowledge_gaps( text=root_cause, triggered_by=ticket_id, day=self.state.day, @@ -1825,6 +1763,21 @@ def _handle_incident(self): timestamp=incident_start_iso, ) + involves_gap = len(detected_gaps) > 0 + gap_areas = [g.domain_hit for g in detected_gaps] + gap_context_str = "" + + if involves_gap: + details = [ + f"{g.departed_name} (owned {g.domain_hit}, {int(g.documented_pct * 100)}% documented)" + for g in detected_gaps + ] + gap_context_str = ( + f"KNOWLEDGE GAP FLAG: This incident touches underdocumented systems. " + f"{' | '.join(details)}. " + f"Resolution may be blocked pending knowledge recovery." + ) + prior = self._recurrence_detector.find_prior_incident( root_cause, self.state.day, ticket_id ) diff --git a/src/genesis.py b/src/genesis.py index abe1c92..c24c75f 100644 --- a/src/genesis.py +++ b/src/genesis.py @@ -16,7 +16,7 @@ LEGACY, ORG_CHART, ) -from memory import Memory +from memory import Memory, SimEvent from agent_factory import make_agent from crewai import Task, Crew @@ -43,6 +43,7 @@ def initialize(config, planner_llm, reset=False): seed_tech_stack(mem, planner_llm) seed_external_sources(mem, planner_llm) seed_crm_accounts(mem) + seed_knowledge_gaps(mem) logger.info("[genesis] ✅ Seeding complete.") return mem @@ -334,6 +335,54 @@ def seed_crm_accounts(mem: Memory): pass +def seed_knowledge_gaps(mem: Memory): + """Embeds skills and logs departure events for pre-simulation employees.""" + if not CONFIG.get("knowledge_gaps"): + return + + logger.info("[genesis] Seeding pre-simulation knowledge gaps...") + + sim_start = datetime.strptime(CONFIG["simulation"]["start_date"], "%Y-%m-%d") + + for gap in CONFIG.get("knowledge_gaps", []): + name = gap["name"] + left_date = gap["left"] + left_dt = datetime.strptime(left_date, "%Y-%m") + departure_day = -(sim_start - left_dt).days + + mem.embed_persona_skills( + name=name, + data={ + "expertise": gap.get("knew_about", []), + "social_role": gap.get("role", "Former Employee"), + }, + dept=gap.get("dept", "Engineering"), + day=departure_day, + timestamp_iso=f"{left_date}-01T09:00:00", + ) + + mem.log_event( + SimEvent( + type="employee_departed", + day=departure_day, + date=f"{left_date}-01", + timestamp=f"{left_date}-01T09:00:00", + actors=[name], + artifact_ids={}, + facts={ + "name": name, + "role": gap.get("role", ""), + "knowledge_domains": gap.get("knew_about", []), + "documented_pct": gap.get("documented_pct", 0.5), + "is_genesis_gap": True, + }, + summary=f"Genesis Gap: {name} ({gap.get('role')}) left Day {departure_day}.", + tags=["employee_departed", "lifecycle", "genesis"], + ) + ) + logger.info(f" [dim]→ Seeded Genesis Gap: {name}[/dim]") + + @staticmethod def _parse_sources(raw: str) -> List[dict]: cleaned = re.sub(r"^```[a-z]*\n?", "", raw.strip()).rstrip("` \n") diff --git a/src/memory.py b/src/memory.py index b562f34..45bbd2d 100644 --- a/src/memory.py +++ b/src/memory.py @@ -483,17 +483,23 @@ def _init_vector_indexes(self): } for coll_name in ["artifacts", "events"]: - if coll_name not in self._db.list_collection_names(): - try: - self._db.create_collection(coll_name) - logger.info(f"[memory] Created collection: {coll_name}") - except Exception: - pass - coll = self._db[coll_name] - existing_indexes = list(coll.list_search_indexes()) - if not any(idx.get("name") == "vector_index" for idx in existing_indexes): + existing = list(coll.list_search_indexes()) + vector_idx = next( + (i for i in existing if i.get("name") == "vector_index"), None + ) + + if vector_idx: + status = vector_idx.get("status") + if status in ("FAILED", "DOES_NOT_EXIST"): + logger.warning( + f"[memory] Index on {coll_name} is {status}. Dropping..." + ) + coll.drop_search_index("vector_index") + vector_idx = None + + if not vector_idx: try: search_index_model = SearchIndexModel( definition=index_definition, @@ -501,9 +507,9 @@ def _init_vector_indexes(self): type="vectorSearch", ) coll.create_search_index(model=search_index_model) - logger.info(f"[memory] Created vector_index on {coll.name}") + logger.info(f"[memory] Created vector_index on {coll_name}") except Exception as e: - logger.error(f"[memory] Could not create index on {coll.name}: {e}") + logger.error(f"[memory] Failed to create index on {coll_name}: {e}") # ─── WRITE ──────────────────────────────── diff --git a/src/normal_day.py b/src/normal_day.py index 3f3c131..47c6738 100644 --- a/src/normal_day.py +++ b/src/normal_day.py @@ -3023,38 +3023,21 @@ def _save_slack( "ts", self._clock.now("system").isoformat() ) - if self._embed_worker: - self._embed_worker.enqueue( - id=thread_id, - type="slack_thread", - title=f"{interaction_type.replace('_', ' ').title()} in #{channel}", - content=full_transcript, - day=self._state.day, - date=date_str, - timestamp=start_timestamp, - metadata={ - "channel": channel, - "interaction_type": interaction_type, - "participants": list({m["user"] for m in messages}), - "message_count": len(messages), - }, - ) - else: - self._mem.embed_artifact( - id=thread_id, - type="slack_thread", - title=f"{interaction_type.replace('_', ' ').title()} in #{channel}", - content=full_transcript, - day=self._state.day, - date=date_str, - timestamp=start_timestamp, - metadata={ - "channel": channel, - "interaction_type": interaction_type, - "participants": list({m["user"] for m in messages}), - "message_count": len(messages), - }, - ) + self._mem.embed_artifact( + id=thread_id, + type="slack_thread", + title=f"{interaction_type.replace('_', ' ').title()} in #{channel}", + content=full_transcript, + day=self._state.day, + date=date_str, + timestamp=start_timestamp, + metadata={ + "channel": channel, + "interaction_type": interaction_type, + "participants": list({m["user"] for m in messages}), + "message_count": len(messages), + }, + ) return slack_path, thread_id @@ -3283,28 +3266,16 @@ def _save_zoom_transcript( "medium": "zoom", } - if self._embed_worker: - self._embed_worker.enqueue( - id=transcript_id, - type="zoom_transcript", - title=f"Zoom: {topic[:80]}", - content=full_text, - day=self._state.day, - date=date_str, - timestamp=meeting_time_iso, - metadata=embed_metadata, - ) - else: - self._mem.embed_artifact( - id=transcript_id, - type="zoom_transcript", - title=f"Zoom: {topic[:80]}", - content=full_text, - day=self._state.day, - date=date_str, - timestamp=meeting_time_iso, - metadata=embed_metadata, - ) + self._mem.embed_artifact( + id=transcript_id, + type="zoom_transcript", + title=f"Zoom: {topic[:80]}", + content=full_text, + day=self._state.day, + date=date_str, + timestamp=meeting_time_iso, + metadata=embed_metadata, + ) logger.info( f" [dim]📹 Zoom transcript saved: {transcript_id} " diff --git a/src/org_lifecycle.py b/src/org_lifecycle.py index c7f9bf3..c05e9fc 100644 --- a/src/org_lifecycle.py +++ b/src/org_lifecycle.py @@ -26,6 +26,7 @@ from __future__ import annotations +from datetime import datetime import logging import json as _json import random @@ -122,7 +123,6 @@ def __init__( self._gap_events: List[KnowledgeGapEvent] = [] self._domains_surfaced: Set[str] = set() - # Build day-keyed lookup tables from config self._scheduled_departures: Dict[int, List[dict]] = {} for dep in self._cfg.get("scheduled_departures", []): self._scheduled_departures.setdefault(dep["day"], []).append(dep) @@ -131,7 +131,27 @@ def __init__( for hire in self._cfg.get("scheduled_hires", []): self._scheduled_hires.setdefault(hire["day"], []).append(hire) - # ─── PUBLIC ─────────────────────────────────────────────────────────────── + sim_start_str = config.get("simulation", {}).get("start_date", "2024-01-01") + sim_start = datetime.strptime(sim_start_str, "%Y-%m-%d") + + for gap in config.get("knowledge_gaps", []): + left_dt = datetime.strptime(gap["left"], "%Y-%m") + day = -(sim_start - left_dt).days + + self._departed.append( + DepartureRecord( + name=gap["name"], + dept=gap.get("dept", "Unknown"), + role=gap.get("role", "Former Employee"), + day=day, + reason="voluntary", + knowledge_domains=gap.get("knew_about", []), + documented_pct=float(gap.get("documented_pct", 0.5)), + peak_stress=50, + edge_snapshot={}, + centrality_at_departure=0.0, + ) + ) def process_departures( self, day: int, date_str: str, state, clock @@ -206,56 +226,141 @@ def scan_for_knowledge_gaps( date_str: str, state, timestamp: str, + similarity_threshold: float = 0.65, ) -> List[KnowledgeGapEvent]: + """ + Detect knowledge gaps using semantic similarity. + + Instead of checking whether a departed employee's domain keyword appears + verbatim in the incident text, we embed the incident text and compare it + against the departed employee's persona_skill artifacts (expertise profile) + and any author_expertise artifacts (topics they wrote about). + + This catches cases where incident terminology differs from the departed + employee's stated expertise — e.g., "auth timeout" matches against + "identity management" because the embeddings are semantically close. + + Args: + text: The incident root cause or description text. + triggered_by: The artifact ID (e.g., JIRA ticket) that surfaced this. + day: Current simulation day. + date_str: Current date as ISO string. + state: Simulation state object. + timestamp: ISO timestamp of the triggering event. + similarity_threshold: Minimum vector similarity score (0–1) to consider + a departed employee's expertise a match. Default 0.65 + is tuned for dotProduct with 1024-dim vectors. + """ found: List[KnowledgeGapEvent] = [] - text_lower = text.lower() + + if not self._departed: + return found + + expert_matches = self._mem.find_expert_by_skill(text, n=20) + + match_scores: Dict[str, float] = {} + for match in expert_matches: + name = match.get("name") + score = match.get("score", 0.0) + if name and score >= similarity_threshold: + if name not in match_scores or score > match_scores[name]: + match_scores[name] = score + for record in self._departed: - for domain in record.knowledge_domains: - if domain.lower() not in text_lower: - continue - gap_key = f"{record.name}:{domain}:{triggered_by}" - if gap_key in self._domains_surfaced: - continue - self._domains_surfaced.add(gap_key) - gap_event = KnowledgeGapEvent( - departed_name=record.name, - domain_hit=domain, - triggered_by=triggered_by, - triggered_on_day=day, - documented_pct=record.documented_pct, - ) - self._gap_events.append(gap_event) - found.append(gap_event) - self._mem.log_event( - SimEvent( - type="knowledge_gap_detected", - timestamp=timestamp, - day=day, - date=date_str, - actors=[record.name], - artifact_ids={"jira": triggered_by}, - facts={ - "departed_employee": record.name, - "gap_areas": [domain], - "triggered_by": triggered_by, - "documented_pct": record.documented_pct, - "days_since_departure": day - record.day, - "escalation_harder": True, - }, - summary=( - f"Knowledge gap: {domain} (owned by ex-{record.name}) " - f"surfaced in {triggered_by}. " - f"~{int(record.documented_pct * 100)}% documented." - ), - tags=["knowledge_gap", "departed_employee"], - ) - ) - logger.info( - f" [yellow]⚠ Knowledge gap:[/yellow] {domain} " - f"(was {record.name}'s) surfaced in {triggered_by}" + if record.name not in match_scores: + continue + + score = match_scores[record.name] + + gap_key = f"{record.name}:semantic:{triggered_by}" + if gap_key in self._domains_surfaced: + continue + self._domains_surfaced.add(gap_key) + + gap_domains = ( + record.knowledge_domains + if record.knowledge_domains + else ["undocumented expertise"] + ) + domain_label = ", ".join(gap_domains) + + gap_event = KnowledgeGapEvent( + departed_name=record.name, + domain_hit=domain_label, + triggered_by=triggered_by, + triggered_on_day=day, + documented_pct=record.documented_pct, + ) + self._gap_events.append(gap_event) + found.append(gap_event) + + self._mem.log_event( + SimEvent( + type="knowledge_gap_detected", + timestamp=timestamp, + day=day, + date=date_str, + actors=[record.name], + artifact_ids={"jira": triggered_by}, + facts={ + "departed_employee": record.name, + "gap_areas": gap_domains, + "triggered_by": triggered_by, + "documented_pct": record.documented_pct, + "days_since_departure": day - record.day, + "escalation_harder": True, + "semantic_score": round(score, 4), + "detection_method": "embedding_similarity", + }, + summary=( + f"Knowledge gap: {domain_label} (owned by ex-{record.name}, " + f"similarity={score:.3f}) surfaced in {triggered_by}. " + f"~{int(record.documented_pct * 100)}% documented." + ), + tags=["knowledge_gap", "departed_employee"], ) + ) + logger.info( + f" [yellow]⚠ Knowledge gap:[/yellow] {domain_label} " + f"(was {record.name}'s, score={score:.3f}) surfaced in {triggered_by}" + ) + return found + def _load_departed_from_log(self, events: List[SimEvent]) -> None: + """ + Reconstructs the _departed registry from the SimEvent log. + Used by backfill scripts that run after the sim completes. + """ + + for e in events: + if e.type != "employee_departed": + continue + name = next(iter(e.actors), None) + if not name: + continue + # Avoid duplicates if called multiple times + if any(r.name == name for r in self._departed): + continue + self._departed.append( + DepartureRecord( + name=name, + day=e.day, + knowledge_domains=e.facts.get("knowledge_domains", []), + documented_pct=e.facts.get("documented_pct", 0.0), + dept=e.facts.get("dept", ""), + role=e.facts.get("role", e.facts.get("dept", "")), + reason=e.facts.get("reason", "voluntary"), + peak_stress=e.facts.get("peak_stress", 50), + edge_snapshot={ + bond[0]: bond[1] for bond in e.facts.get("strongest_bonds", []) + }, + ) + ) + logger.info( + f"[lifecycle] Loaded {len(self._departed)} departed employee(s) from event log." + ) + def get_roster_context(self) -> str: lines: List[str] = [] for d in self._departed[-3:]: diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py index f23e394..0807cfc 100644 --- a/tests/test_lifecycle.py +++ b/tests/test_lifecycle.py @@ -482,11 +482,6 @@ def test_departure_record_stored_on_state(lifecycle, mock_clock): assert "redis-cache" in state.departed_employees["Bob"]["knew_about"] -# ───────────────────────────────────────────────────────────────────────────── -# 5. KNOWLEDGE GAP SCANNING -# ───────────────────────────────────────────────────────────────────────────── - - def test_knowledge_gap_scan_detects_domain_hit(lifecycle, mock_clock): """ scan_for_knowledge_gaps must emit a knowledge_gap_detected SimEvent when @@ -495,7 +490,6 @@ def test_knowledge_gap_scan_detects_domain_hit(lifecycle, mock_clock): mgr, gd, org_chart, all_names, state = lifecycle state.active_incidents = [] - # First, register a departure with known domains dep_cfg = { "name": "Bob", "reason": "voluntary", @@ -506,30 +500,35 @@ def test_knowledge_gap_scan_detects_domain_hit(lifecycle, mock_clock): } mgr._scheduled_departures = {3: [dep_cfg]} mgr.process_departures(day=3, date_str="2026-01-03", state=state, clock=mock_clock) - mgr._mem.log_event.reset_mock() - # Now trigger a scan with text that mentions the domain - gaps = mgr.scan_for_knowledge_gaps( - text="Root cause: auth-service JWT validation failing after config change.", - triggered_by="ORG-400", - day=5, - date_str="2026-01-05", - state=state, - timestamp=mock_clock, - ) + with patch.object(mgr._mem, "find_expert_by_skill") as mock_search: + mock_search.return_value = [ + {"name": "Bob", "score": 0.85, "dept": "Engineering"} + ] + mgr._mem.log_event.reset_mock() + + text = "auth-service and redis-cache are failing." + gaps = mgr.scan_for_knowledge_gaps( + text=text, + triggered_by="ORG-400", + day=5, + date_str="2026-01-05", + state=state, + timestamp="2026-01-05T10:00:00", + ) - assert len(gaps) == 1 - assert gaps[0].domain_hit == "auth-service" - assert gaps[0].departed_name == "Bob" - assert gaps[0].triggered_by == "ORG-400" - assert gaps[0].documented_pct == 0.25 + assert len(gaps) == 1 - gap_events = [ - call.args[0] - for call in mgr._mem.log_event.call_args_list - if call.args[0].type == "knowledge_gap_detected" - ] - assert len(gap_events) == 1 + assert gaps[0].domain_hit == "auth-service, redis-cache" + assert gaps[0].departed_name == "Bob" + + gap_events = [ + c.args[0] + for c in mgr._mem.log_event.call_args_list + if c.args[0].type == "knowledge_gap_detected" + ] + assert len(gap_events) == 1 + assert gap_events[0].facts["gap_areas"] == ["auth-service", "redis-cache"] def test_knowledge_gap_scan_deduplicates(lifecycle, mock_clock): @@ -550,32 +549,48 @@ def test_knowledge_gap_scan_deduplicates(lifecycle, mock_clock): } mgr._scheduled_departures = {3: [dep_cfg]} mgr.process_departures(day=3, date_str="2026-01-03", state=state, clock=mock_clock) - mgr._mem.log_event.reset_mock() + with patch.object(mgr._mem, "find_expert_by_skill") as mock_search: + mock_search.return_value = [ + {"name": "Bob", "score": 0.9, "dept": "Engineering"} + ] + mgr._mem.log_event.reset_mock() - text = "redis-cache connection pool exhausted" - mgr.scan_for_knowledge_gaps( - text=text, - triggered_by="ORG-401", - day=5, - date_str="2026-01-05", - state=state, - timestamp=mock_clock, - ) - mgr.scan_for_knowledge_gaps( - text=text, - triggered_by="ORG-402", - day=6, - date_str="2026-01-06", - state=state, - timestamp=mock_clock, - ) + text = "redis-cache connection pool exhausted" - gap_events = [ - call.args[0] - for call in mgr._mem.log_event.call_args_list - if call.args[0].type == "knowledge_gap_detected" - ] - assert len(gap_events) == 2 + mgr.scan_for_knowledge_gaps( + text=text, + triggered_by="ORG-401", + day=5, + date_str="x", + state=state, + timestamp="t", + ) + + mgr.scan_for_knowledge_gaps( + text=text, + triggered_by="ORG-401", + day=5, + date_str="x", + state=state, + timestamp="t", + ) + + mgr.scan_for_knowledge_gaps( + text=text, + triggered_by="ORG-402", + day=6, + date_str="y", + state=state, + timestamp="t", + ) + + gap_events = [ + call.args[0] + for call in mgr._mem.log_event.call_args_list + if call.args[0].type == "knowledge_gap_detected" + ] + + assert len(gap_events) == 2 def test_knowledge_gap_scan_no_false_positives(lifecycle, mock_clock):