"""Nightly reflection engine — the proactive brain (runs on a Railway cron). Reads the shared memory graph over the memory API, asks DeepSeek V4 Pro to reason over the whole thing — spotting OPPORTUNITIES (non-obvious money/growth moves), OBSERVATIONS (patterns or risks Ahmed might miss), and NUDGES (open commitments needing action) — and posts each finding back as an :Insight so any device can surface it. No agent, no Claude Code: just DeepSeek + HTTP. Dependency-free. On top of the insight pass it also, each night (ORDER matters): - Pass 2: writes a tight current-state summary onto the top entities. - Pass 3: extracts Ahmed's open commitments into tasks. - Pass 4: consolidates near-duplicate memories (retires the stale copies). - Pass 5: writes a psychological "read" onto the top people/companies. - Pass 6: refines Ahmed's north-star value model. - Pass 7: entity hygiene — auto-dedup entities, adjudicate the ambiguous band, and flag mistyped "thing" entities (log-only, no retype endpoint). - Pass 8: pattern mining — recurring routines/cycles saved as kind="pattern". - Pass 9: synthesis — higher-level conclusions saved as kind="synthesis". - Pass 10: core block — refresh the pinned ~200-word profile (LAST, so it sees the night's cleanup) and POST /core. - Pass 11: archive sweep — retire stale, low-value, isolated observations/events. The run is bracketed by a HEARTBEAT (POST /reflection_ran started → ok/error) so a silent death is visible — this is why reflection quietly stopped before. Every generative pass is threaded with Ahmed's stated preferences/feedback so the whole reflection aligns to what he's told Jarvis to do (or stop doing). Newer memory-service endpoints (added by a parallel effort) are called through `mem_opt`, which logs-and-continues on a 404 so an un-deployed endpoint never takes the whole run down: /reflection_ran, /core, /core_block, /entities/dedup_backfill, /entities/merge. Memories may also now carry importance / access_count / last_accessed / valid_from|to / refers_to — all read defensively (they may be absent). Env: MEMORY_API_URL, MEMORY_API_KEY, DEEPSEEK_API_KEY (optional) REFLECT_MODEL (default deepseek-v4-pro), MEMORY_GROUP, REFLECT_MAX_MEMORIES (context cap, default 800; paginated), REFLECT_DRY=1 to print findings instead of storing them (applies to EVERY pass, including the heartbeat and all the new passes). """ import datetime as _dt import json import os import re import urllib.error import urllib.parse import urllib.request MEM = os.environ["MEMORY_API_URL"].rstrip("/") MEM_KEY = os.environ.get("MEMORY_API_KEY", "") GROUP = os.environ.get("MEMORY_GROUP", "ahmed") DS_URL = os.environ.get("DEEPSEEK_URL", "https://api.deepseek.com").rstrip("/") DS_KEY = os.environ["DEEPSEEK_API_KEY"] MODEL = os.environ.get("REFLECT_MODEL", "deepseek-v4-pro") DRY = os.environ.get("REFLECT_DRY", "0") == "1" try: MAX_MEMORIES = max(1, int(os.environ.get("REFLECT_MAX_MEMORIES", "800"))) except ValueError: MAX_MEMORIES = 800 def mem(method, path, body=None, timeout=60): data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(MEM + path, data=data, method=method) if MEM_KEY: req.add_header("Authorization", f"Bearer {MEM_KEY}") req.add_header("Content-Type", "application/json") with urllib.request.urlopen(req, timeout=timeout) as r: raw = r.read() return json.loads(raw) if raw else {} def mem_opt(method, path, body=None, timeout=60): """Like mem() but NEVER raises — returns None (and logs) on any error. For the newer endpoints a parallel effort is adding (/reflection_ran, /core, /core_block, /entities/*). If one 404s because it isn't deployed yet we just log and keep going so the rest of the reflection still runs.""" try: return mem(method, path, body, timeout) except urllib.error.HTTPError as e: print(f" [mem] {method} {path} -> HTTP {e.code} (skipping)", flush=True) except Exception as e: # noqa: BLE001 print(f" [mem] {method} {path} failed: {e} (skipping)", flush=True) return None def heartbeat(status, note=""): """Stamp POST /reflection_ran so a silent death is visible on the dashboard. NEVER raises — a failing heartbeat must not take the run down. Honours REFLECT_DRY (prints instead of writing).""" note = str(note or "")[:1800] if DRY: print(f" [heartbeat] {status}: {note}", flush=True) return try: mem("POST", "/reflection_ran", {"status": status, "note": note, "group": GROUP}) except Exception as e: # noqa: BLE001 print(f" [heartbeat] {status} post failed: {e}", flush=True) def deepseek(prompt, max_tokens=6000): body = json.dumps({"model": MODEL, "temperature": 0.4, "stream": False, "max_tokens": max_tokens, "messages": [{"role": "user", "content": prompt}]}).encode() req = urllib.request.Request( DS_URL + "/chat/completions", data=body, headers={"Content-Type": "application/json", "Authorization": f"Bearer {DS_KEY}"}) with urllib.request.urlopen(req, timeout=300) as r: o = json.loads(r.read()) return ((o.get("choices") or [{}])[0].get("message") or {}).get("content", "").strip() PROMPT = """You are Jarvis's reflection engine — a sharp chief-of-staff reasoning \ over everything Ahmed's assistant knows about his work and life. Below is his \ entire memory. Think hard and surface only GENUINELY valuable, non-obvious \ findings a great advisor would raise unprompted. Produce up to 6 findings across three kinds: - opportunity: a non-obvious move that could make money or advance a goal (a \ client/partner fit, an intro to make, an upsell, a reusable asset). - observation: a pattern or risk he might miss (leads going cold, over-reliance \ on one client, a bottleneck, a contradiction). - nudge: an open commitment or goal that needs action soon. Be SPECIFIC — name the people, companies, and numbers. Skip anything obvious, \ generic, or low-value. Quality over quantity; 0 findings is fine if nothing is \ worth his attention. Output ONLY a JSON array: [{{"type":"opportunity|observation|nudge","title":"short label","detail":"1-2 \ specific sentences","entities":["names"]}}] ENTITIES (by how often they come up): {ents} FACTS: {facts} JSON:""" SUMMARY_PROMPT = """You are Jarvis's memory summarizer. Below are all the facts \ the assistant knows about "{name}". Write a TIGHT current-state summary in 2-4 \ sentences: who or what they are, current status, key numbers, the latest \ development, and any open items. Plain prose only — no markdown, no bullet \ points, no preamble, no surrounding quotes. If the facts are thin, keep it to \ one honest sentence. FACTS about {name}: {facts} SUMMARY:""" GOALS_PROMPT = """You are Jarvis's chief-of-staff. Scan ALL of Ahmed's memory \ below and extract his OPEN, actionable COMMITMENTS — concrete things HE said he \ WILL DO, ideally with a deadline (e.g. "call Bloom Well by end of July", "send \ Al Temimi the proposal Thursday"). Be CONSERVATIVE: only real commitments he actually made and that are still open \ — NOT vague intentions, wishes, ideas, or things already done. If nothing \ qualifies, return []. Output ONLY a JSON array, each item an imperative task: [{{"text":"imperative task phrased as an action","due":"ISO datetime or empty \ string"}}] FACTS: {facts} JSON:""" CONSOLIDATE_PROMPT = """You are Jarvis's memory-consolidation engine. Below are \ Ahmed's memory facts, each on its own line as "id: text". Find GROUPS of facts \ that are near-duplicates — they say essentially the SAME thing (e.g. three \ separate "Ahmed is at home" facts, or two phrasings of the same single fact). Only group GENUINE duplicates. Do NOT group facts that are merely related, on \ the same topic, sequential, or complementary — those must stay separate. When \ unsure, leave them out. Returning [] is fine. Output ONLY a JSON array of groups; each group is an array of ids that duplicate \ each other: [["id1","id2"], ["id7","id8","id9"]] FACTS: {facts} JSON:""" READS_PROMPT = """You are Jarvis's read-the-room engine. Below are all the facts \ the assistant knows about "{name}". Produce a TIGHT psychological READ, in 2-4 \ sentences, that models THEM for Ahmed's benefit — not a summary of events but an \ assessment of the person/company: what they seem to WANT and VALUE, how they make \ decisions, the current temperature of Ahmed's relationship with them, and how \ Ahmed should APPROACH them next. Ground every claim in the facts and flag \ uncertainty ("seems", "likely") — do NOT invent traits or details that aren't \ there. If the facts are too thin to read them, say so honestly in one sentence. \ Plain prose only — no markdown, no bullet points, no preamble, no surrounding \ quotes. FACTS about {name}: {facts} READ:""" VALUES_PROMPT = """You are Jarvis's model of Ahmed himself. From everything the \ assistant knows about him below, infer his NORTH STAR: his real priorities, what \ he chases, what he avoids or dislikes, and how he weighs tradeoffs (e.g. margin \ vs volume, growth vs quick cash, time vs money, risk vs safety, autonomy vs \ delegation). Write a tight profile — 4 to 8 sentences or short lines of plain prose. No \ markdown, no bullet points, no preamble, no surrounding quotes. If a PRIOR \ PROFILE is given below, this is an UPDATE that REFINES it: keep what still holds, \ adjust only what the facts now contradict, and fold in what's genuinely new — do \ NOT wipe it and start fresh. {prior} FACTS about Ahmed: {facts} PROFILE:""" MERGE_ADJUDICATE_PROMPT = """You are Jarvis's entity-resolution judge. Each PAIR \ below is two memory entities the system suspects are the SAME real-world entity \ (person, company, place, or product), with a similarity score and each side's \ facts. For every pair decide: are they truly the SAME entity that should be \ MERGED, or DIFFERENT and kept apart? Be CONSERVATIVE — only merge when you're confident they name the identical real \ thing. Two different people who share a first name, a company vs. its founder, or \ a brand vs. one of its products are DIFFERENT. When unsure, keep. Output ONLY a JSON array, one verdict per pair (use "keep" only when merging — it \ says which side's name to keep): [{{"pair":1,"verdict":"merge|keep","keep":"A|B"}}] PAIRS: {pairs} JSON:""" THING_RETYPE_PROMPT = """You are Jarvis's entity-type auditor. Each entity below \ is currently typed as a generic "thing", but its facts may reveal it's really a \ PERSON, COMPANY, PLACE, or PRODUCT. For each, only if the facts CLEARLY indicate \ a more specific type, say so; otherwise leave it out. Output ONLY a JSON array (include ONLY entities that should be retyped): [{{"name":"...","suggested":"person|company|place|product"}}] ENTITIES: {ents} JSON:""" PATTERN_PROMPT = """You are Jarvis's pattern miner. Below are Ahmed's memories, \ each as `id [date | re EVENTDATE]: text` — `date` is when it was recorded, \ `re EVENTDATE` (when present) is the real-world date it refers to. Mine RECURRING \ patterns: routines, cycles, and behaviours that REPEAT across time, each backed by \ specific evidence memories. Good patterns: "Ahmed ships a feature then immediately starts the next with no \ break", "the scraper wedges roughly every N days", "Ahmed reworks Jarvis's voice \ stack every couple of weeks". Only report a pattern supported by at least TWO \ memories; cite their ids. Skip one-offs. Do NOT repeat anything in ALREADY-KNOWN. Output ONLY a JSON array (MAX 5): [{{"text":"Ahmed does X every Y (N times since DATE)","evidence":["id1","id2"]}}] ALREADY-KNOWN PATTERNS (do NOT repeat): {known} MEMORIES: {facts} JSON:""" SYNTHESIS_PROMPT = """You are Jarvis's synthesizer. Read Ahmed's recent memories \ below (each `id: text`) and infer 3-5 HIGHER-LEVEL CONCLUSIONS that logically \ FOLLOW from them but are NOT directly stated in any single memory — the \ connect-the-dots read a sharp analyst would draw. Each conclusion must be \ grounded in specific memories; cite their ids. Skip the obvious and anything in \ ALREADY-KNOWN. Output ONLY a JSON array (MAX 5): [{{"text":"the higher-level conclusion","sources":["id1","id2"]}}] ALREADY-KNOWN (do NOT repeat): {known} MEMORIES: {facts} JSON:""" CORE_PROMPT = """You are Jarvis's profile writer. Write a PINNED CORE PROFILE of \ Ahmed that is injected into EVERY Jarvis prompt — so it must be DENSE, CURRENT, \ and 200 WORDS MAX. Cover, in tight prose: who Ahmed is; each ACTIVE project with a \ one-line current state; the key people and companies and their role to him; his \ current priorities; and his standing preferences. CRITICAL: include NOTHING stale or superseded. If a newer fact overrides an older \ one, keep ONLY the newer. Drop anything finished, abandoned, or out of date. No \ markdown, no bullet symbols, no preamble, no surrounding quotes — just prose \ (short lines are fine). HARD LIMIT 200 words. {prior}ENTITY SUMMARIES (current state, refreshed tonight): {summaries} HIGH-SIGNAL FACTS (importance / recency / usage ranked): {facts} CORE PROFILE:""" ARCHIVE_SUMMARY_PROMPT = """You are Jarvis's memory archiver. Each CLUSTER below \ is a group of old, low-value memories about to be retired. For each, write ONE \ short factual sentence that preserves the durable gist so nothing important is \ lost. Plain prose, no markdown, no preamble. Output ONLY a JSON array: [{{"cluster":1,"text":"one-sentence summary of the cluster"}}] CLUSTERS: {clusters} JSON:""" def _fetch_memories(cap): """Page through /memories up to `cap` (the service caps each page at ~200, so a single request can't honour a big cap — we page by offset). Newest-first.""" out, offset = [], 0 while len(out) < cap: want = min(200, cap - len(out)) res = mem("GET", "/memories?" + urllib.parse.urlencode( {"group": GROUP, "limit": want, "offset": offset})) batch = [f for f in (res or {}).get("memories", []) if isinstance(f, dict)] if not batch: break out.extend(batch) if len(batch) < want: # last page break offset += len(batch) return out[:cap] def build_context(): facts = _fetch_memories(MAX_MEMORIES) # Defensive value-aware trim: pagination already bounds the corpus to the cap # (newest-first), but if the API ever over-returns we keep the highest-value # by importance → access_count → recency rather than dropping arbitrarily. if len(facts) > MAX_MEMORIES: facts = sorted(facts, key=_mem_rank, reverse=True)[:MAX_MEMORIES] graph = mem("GET", "/graph?" + urllib.parse.urlencode({"group": GROUP})) ents = {} for n in graph.get("nodes", []): if str(n.get("id", "")).startswith("ent:"): ents[n["id"]] = {"name": n.get("text", ""), "kind": str(n.get("kind", "")).replace("entity:", ""), "deg": 0} for e in graph.get("edges", []): if e.get("type") == "mentions" and e.get("to") in ents: ents[e["to"]]["deg"] += 1 # Tone: HOW Ahmed said a fact (pace/pitch/laughter, captured on-device) is # appended so the ToM "reads" and value-model passes can weigh delivery, not # just words — e.g. a plan stated flatly/sarcastically vs. one said urgently. facts_txt = "\n".join( f"- {f.get('fact', '')}" + (f" [said {f['tone']}]" if f.get("tone") else "") for f in facts) ents_txt = "\n".join( f"- {e['name']} ({e['kind']}, in {e['deg']} facts)" for e in sorted(ents.values(), key=lambda x: -x["deg"]) if e["deg"] > 0) return facts_txt, ents_txt, facts, graph def _mem_rank(m): """Sort key (use reverse=True) preferring high-importance, then frequently- accessed, then recent. All three fields are OPTIONAL (a parallel effort is adding them; /memories doesn't return them yet) so they default safely — until then this degrades gracefully to recency order.""" imp = m.get("importance") imp = float(imp) if isinstance(imp, (int, float)) else 4.0 ac = m.get("access_count") ac = float(ac) if isinstance(ac, (int, float)) else 0.0 return (imp, ac, str(m.get("created_at") or "")) def _parse_date(s): """First-10-chars ISO date -> datetime.date, or None if unparseable.""" try: return _dt.date.fromisoformat(str(s or "")[:10]) except ValueError: return None def _json_array(raw): """Pull the first JSON array out of a model reply; [] on any failure.""" m = re.search(r"\[.*\]", str(raw or ""), re.DOTALL) if not m: return [] try: val = json.loads(m.group(0)) return val if isinstance(val, list) else [] except Exception: # noqa: BLE001 return [] def _ts_listing(facts, cap=400): """`id [date | re EVENTDATE]: text` lines — timestamps in, for the pattern miner. Capped to bound token spend.""" lines = [] for f in facts[:cap]: fid, txt = f.get("id"), str(f.get("fact", "")).strip() if fid is None or not txt: continue stamp = str(f.get("created_at") or "")[:10] refers = str(f.get("refers_to") or "").strip()[:10] if refers: stamp += f" | re {refers}" lines.append(f"{fid} [{stamp}]: {txt}") return "\n".join(lines) def _id_listing(facts, cap=200): """`id: text` lines (newest-first slice) for the synthesis pass.""" lines = [] for f in facts[:cap]: fid, txt = f.get("id"), str(f.get("fact", "")).strip() if fid is not None and txt: lines.append(f"{fid}: {txt}") return "\n".join(lines) def _entity_facts(name, cap=6): """Up to `cap` fact texts for an entity by name (memory API only, no tokens).""" if not name: return [] try: res = mem("GET", "/entity?" + urllib.parse.urlencode( {"name": name, "group": GROUP})) except Exception: # noqa: BLE001 return [] out = [] for ent in (res or {}).get("entities", []): for f in ent.get("facts", []): t = str(f.get("fact", "")).strip() if t: out.append(t) return out[:cap] def _clean_text(s): """Trim a model reply down to plain prose (drop code fences / wrapping quotes).""" s = str(s or "").strip() if s.startswith("```"): s = re.sub(r"^```[a-zA-Z]*\n?", "", s) s = re.sub(r"\n?```$", "", s).strip() if len(s) >= 2 and s[0] in "\"'" and s[-1] == s[0]: s = s[1:-1].strip() return s def feedback_block(): """Ahmed's stated preferences (kind=feedback + kind=preference) as a prompt block. Returns a short "RESPECT THESE" block to PREPEND onto every generative pass so the whole reflection aligns to what he's told Jarvis. Returns "" when he's stated none (a safe no-op to prepend anywhere). Computed once per run. """ items = [] for kind in ("feedback", "preference"): try: res = mem("GET", "/memories?" + urllib.parse.urlencode( {"group": GROUP, "kind": kind, "limit": 30})) for m in res.get("memories", []): txt = str(m.get("fact", "")).strip() if txt: items.append(txt) except Exception: # noqa: BLE001 continue seen, uniq = set(), [] for t in items: k = t.lower() if k in seen: continue seen.add(k) uniq.append(t) if not uniq: return "" lines = "\n".join(f"- {t}" for t in uniq) return ("\n\nAHMED'S STATED PREFERENCES — RESPECT THESE (things he's told you " "to do or stop):\n" + lines + "\n") def pass_summaries(feedback=""): """Pass 2 — write a current-state summary onto the entities that NEED one. TOUCHED-FIRST selection (freshness scales with daily activity, not with total graph size): every entity mentioned by a fact from the last 48h is summarized first — a client signed today gets a fresh summary TONIGHT even if some old heavyweight out-mentions it — then remaining slots fill by all-time mention count. Skip <2 facts; upserts via POST /summary.""" graph = mem("GET", "/graph?" + urllib.parse.urlencode({"group": GROUP})) # memory ids from the last 48h (page 1 newest-first is plenty for a day) recent_ids = set() try: cutoff = (_dt.datetime.now(_dt.timezone.utc) - _dt.timedelta(hours=48)).isoformat() for m in mem("GET", "/memories?" + urllib.parse.urlencode( {"group": GROUP, "limit": 200})).get("memories", []): if str(m.get("created_at", "")) >= cutoff: recent_ids.add(str(m.get("id", ""))) except Exception as e: # noqa: BLE001 — degrade to pure top-degree print(f" [summaries] recent-window fetch failed: {str(e)[:80]}", flush=True) ents = {} for n in graph.get("nodes", []): nid = str(n.get("id", "")) if nid.startswith("ent:"): ents[nid] = {"name": str(n.get("text", "")).strip(), "deg": 0, "touched": False} for e in graph.get("edges", []): if e.get("type") == "mentions" and e.get("to") in ents: ents[e["to"]]["deg"] += 1 if str(e.get("from", "")) in recent_ids: ents[e["to"]]["touched"] = True ranked = sorted(ents.values(), key=lambda x: (-int(x["touched"]), -x["deg"])) top = [e for e in ranked if e["deg"] >= 2 and e["name"]][:15] n_touch = sum(1 for e in top if e["touched"]) print(f" [summaries] {n_touch} touched-recently + " f"{len(top) - n_touch} top-degree selected", flush=True) done = 0 for e in top: name = e["name"] try: res = mem("GET", "/entity?" + urllib.parse.urlencode( {"name": name, "group": GROUP})) ent_facts = [] for ent in res.get("entities", []): for f in ent.get("facts", []): txt = str(f.get("fact", "")).strip() if txt: ent_facts.append(txt) if not ent_facts: continue block = "\n".join(f"- {x}" for x in ent_facts) summary = _clean_text(deepseek( feedback + SUMMARY_PROMPT.format(name=name, facts=block), max_tokens=2500)) # V4 reasons before answering — headroom if len(summary) < 10: continue if DRY: print(f" [summary] {name}: {summary}", flush=True) done += 1 continue mem("POST", "/summary", {"name": name, "text": summary, "group": GROUP}) done += 1 except Exception: # noqa: BLE001 continue return done def _dup_task(text, existing): """True if `text` is already represented in `existing` (list of lowercased task texts).""" t = text.lower().strip() if not t: return True tw = {w for w in re.findall(r"\w+", t) if len(w) > 3} for e in existing: e = str(e).lower().strip() if not e: continue if t in e or e in t: return True ew = {w for w in re.findall(r"\w+", e) if len(w) > 3} if tw and ew and len(tw & ew) / min(len(tw), len(ew)) >= 0.6: return True return False def pass_goals(facts_txt, feedback=""): """Pass 3 — extract open commitments and create tasks (deduped vs open tasks).""" raw = deepseek(feedback + GOALS_PROMPT.format(facts=facts_txt)) m = re.search(r"\[.*\]", raw, re.DOTALL) try: goals = json.loads(m.group(0)) if m else [] except Exception: # noqa: BLE001 goals = [] goals = [g for g in goals if isinstance(g, dict)] try: open_tasks = mem("GET", "/tasks?" + urllib.parse.urlencode( {"group": GROUP})).get("tasks", []) except Exception: # noqa: BLE001 open_tasks = [] existing = [str(t.get("text", "")) for t in open_tasks if isinstance(t, dict)] created = 0 for g in goals: text = str(g.get("text", "")).strip() due = str(g.get("due", "")).strip() if len(text) < 5 or _dup_task(text, existing): continue if DRY: print(f" [goal] {text} (due {due or 'none'})", flush=True) existing.append(text) created += 1 continue body = {"text": text, "group": GROUP} if due: body["due"] = due try: mem("POST", "/task", body) existing.append(text) created += 1 except Exception: # noqa: BLE001 continue return created def pass_consolidate(facts): """Pass 4 — retire near-duplicate memories, keeping the newest of each group. Reuses the corpus build_context already fetched (no extra memory read).""" facts = [f for f in facts if isinstance(f, dict) and f.get("id") is not None and f.get("fact")] if len(facts) < 2: return 0 by_id = {str(f["id"]): f for f in facts} listing = "\n".join(f"{f['id']}: {f['fact']}" for f in facts) raw = deepseek(CONSOLIDATE_PROMPT.format(facts=listing)) m = re.search(r"\[.*\]", raw, re.DOTALL) try: groups = json.loads(m.group(0)) if m else [] except Exception: # noqa: BLE001 groups = [] retired = 0 for grp in groups: if not isinstance(grp, list): continue ids = [str(x) for x in grp if str(x) in by_id] ids = list(dict.fromkeys(ids)) # de-dup, keep order if len(ids) < 2: continue # keep the newest by created_at; retire the rest ids.sort(key=lambda i: str(by_id[i].get("created_at", ""))) keep, drop = ids[-1], ids[:-1] if DRY: print(f" [consolidate] keep {keep} retire {', '.join(drop)}", flush=True) retired += len(drop) continue for i in drop: try: mem("POST", "/retire", {"id": by_id[i]["id"], "group": GROUP}) retired += 1 except Exception: # noqa: BLE001 continue return retired def pass_reads(feedback=""): """Pass 5 — write a psychological READ onto the top ~8 people/companies. Like pass_summaries, but scoped to entity nodes whose type is person or company, picked by incoming `mentions` count. For each it pulls the entity's facts and asks DeepSeek to model what they want/value and how Ahmed should approach them, then upserts via POST /read. Models THEM for Ahmed's benefit. """ graph = mem("GET", "/graph?" + urllib.parse.urlencode({"group": GROUP})) ents = {} for n in graph.get("nodes", []): nid = str(n.get("id", "")) kind = str(n.get("kind", "")) if nid.startswith("ent:") and ( kind.startswith("entity:person") or kind.startswith("entity:company")): ents[nid] = {"name": str(n.get("text", "")).strip(), "deg": 0} for e in graph.get("edges", []): if e.get("type") == "mentions" and e.get("to") in ents: ents[e["to"]]["deg"] += 1 top = [e for e in sorted(ents.values(), key=lambda x: -x["deg"]) if e["deg"] >= 2 and e["name"]][:8] done = 0 for e in top: name = e["name"] try: res = mem("GET", "/entity?" + urllib.parse.urlencode( {"name": name, "group": GROUP})) ent_facts = [] for ent in res.get("entities", []): for f in ent.get("facts", []): txt = str(f.get("fact", "")).strip() if txt: ent_facts.append(txt) if len(ent_facts) < 2: continue block = "\n".join(f"- {x}" for x in ent_facts) read = _clean_text(deepseek( feedback + READS_PROMPT.format(name=name, facts=block), max_tokens=900)) # V4 reasons before answering — give headroom if len(read) < 10: continue if DRY: print(f" [read] {name}: {read}", flush=True) done += 1 continue mem("POST", "/read", {"name": name, "text": read, "group": GROUP}) done += 1 except Exception: # noqa: BLE001 continue return done def pass_values(facts_txt, feedback=""): """Pass 6 — infer/refine Ahmed's north-star value model (singleton /values). One DeepSeek call over all of Ahmed's facts + his existing /values text (if any) + his stated preferences. Refines the prior profile rather than wiping it, then upserts via POST /values. Returns True if a profile was written. """ prior = "" try: cur = mem("GET", "/values?" + urllib.parse.urlencode({"group": GROUP})) prior_txt = str(cur.get("text", "")).strip() if prior_txt: prior = ("\nPRIOR PROFILE (refine this, don't wipe it):\n" + prior_txt + "\n") except Exception: # noqa: BLE001 prior = "" text = _clean_text(deepseek( feedback + VALUES_PROMPT.format(prior=prior, facts=facts_txt), max_tokens=2000)) # big input (all facts) + V4 reasoning needs room if len(text) < 10: return False if DRY: print(f" [values] {text}", flush=True) return True try: mem("POST", "/values", {"text": text, "group": GROUP}) except Exception: # noqa: BLE001 return False return True def _pair_fields(p): """Best-effort extraction of (key_a, name_a, key_b, name_b, score) from an ambiguous-pair record whose exact shape the parallel service owns. Tolerates both flat (*_key) and nested (a/b objects) layouts; None if unusable.""" if not isinstance(p, dict): return None a = p.get("a") if isinstance(p.get("a"), dict) else {} b = p.get("b") if isinstance(p.get("b"), dict) else {} key_a = (p.get("a_key") or p.get("key_a") or p.get("keep_key") or a.get("key") or a.get("id") or a.get("name")) key_b = (p.get("b_key") or p.get("key_b") or p.get("merge_key") or b.get("key") or b.get("id") or b.get("name")) if not key_a or not key_b: return None name_a = (p.get("a_name") or p.get("name_a") or a.get("name") or a.get("text") or key_a) name_b = (p.get("b_name") or p.get("name_b") or b.get("name") or b.get("text") or key_b) score = p.get("score") or p.get("similarity") or p.get("sim") or 0 return (str(key_a), str(name_a), str(key_b), str(name_b), score) def _flag_thing_retypes(feedback=""): """Batch-ask DeepSeek which generic "thing" entities are really a person/ company/place/product. LOG-ONLY — there is no retype endpoint — so it just returns a short note string that lands in the run summary. ONE DeepSeek call.""" graph = mem("GET", "/graph?" + urllib.parse.urlencode({"group": GROUP})) things = {} for n in graph.get("nodes", []): nid = str(n.get("id", "")) if nid.startswith("ent:") and str(n.get("kind", "")) == "entity:thing": name = str(n.get("text", "")).strip() if name: things[nid] = {"name": name, "deg": 0} for e in graph.get("edges", []): if e.get("type") == "mentions" and e.get("to") in things: things[e["to"]]["deg"] += 1 top = [t for t in sorted(things.values(), key=lambda x: -x["deg"]) if t["deg"] >= 2][:20] if not top: return "" blocks = [] for t in top: fx = "; ".join(_entity_facts(t["name"], 5)) or "(no facts)" blocks.append(f'- "{t["name"]}": {fx}') raw = deepseek(feedback + THING_RETYPE_PROMPT.format(ents="\n".join(blocks)), max_tokens=2500) # V4 reasoning headroom sugg = [] for v in _json_array(raw): if not isinstance(v, dict): continue name = str(v.get("name", "")).strip() s = str(v.get("suggested", "")).strip().lower() if name and s in ("person", "company", "place", "product"): sugg.append(f"{name}->{s}") if not sugg: return "" note = "retype: " + ", ".join(sugg[:20]) print(f" [retype] {note}", flush=True) return note def pass_entity_hygiene(feedback=""): """Pass 7 — entity hygiene. (a) POST /entities/dedup_backfill (non-dry, or dry in REFLECT_DRY): the service auto-merges the confident (>=0.95) pairs and returns an 'ambiguous' 0.80-0.95 band for us to judge. (b) ONE batched DeepSeek call adjudicates the WHOLE band → POST /entities/merge for the pairs it confirms. (c) Flags mistyped 'thing' entities (log-only) via _flag_thing_retypes. Returns (merges_done, retype_note). Missing endpoints 404 → (0, "").""" res = mem_opt("POST", "/entities/dedup_backfill", {"dry_run": DRY, "group": GROUP}) merges = 0 if res is not None: band = res.get("ambiguous") or res.get("pairs") or [] pairs = [] for p in (band if isinstance(band, list) else [])[:20]: pf = _pair_fields(p) if pf: pairs.append(pf) if pairs: blocks = [] for i, (ka, na, kb, nb, sc) in enumerate(pairs, 1): fa = "; ".join(_entity_facts(na, 4)) or "(no facts)" fb = "; ".join(_entity_facts(nb, 4)) or "(no facts)" blocks.append(f'Pair {i} (score {sc}): A="{na}" [{fa}] | ' f'B="{nb}" [{fb}]') raw = deepseek(feedback + MERGE_ADJUDICATE_PROMPT.format( pairs="\n".join(blocks)), max_tokens=3000) # V4 reasoning room for v in _json_array(raw): if not isinstance(v, dict): continue try: idx = int(v.get("pair", 0)) - 1 except (TypeError, ValueError): continue if not (0 <= idx < len(pairs)): continue if str(v.get("verdict", "")).strip().lower() != "merge": continue ka, na, kb, nb, sc = pairs[idx] keep_side = str(v.get("keep", "A")).strip().upper() keep_key, merge_key = (kb, ka) if keep_side == "B" else (ka, kb) if DRY: print(f" [entity-merge] keep {keep_key} <- {merge_key}", flush=True) merges += 1 continue if mem_opt("POST", "/entities/merge", {"keep_key": keep_key, "merge_keys": [merge_key], "group": GROUP}) is not None: merges += 1 retype_note = "" try: retype_note = _flag_thing_retypes(feedback) except Exception as e: # noqa: BLE001 print(f" [retype] failed: {e}", flush=True) return merges, retype_note def _existing_kind(kind, limit=100): """Fact texts of memories already stored under `kind` (dedup guard + context).""" out = [] res = mem_opt("GET", "/memories?" + urllib.parse.urlencode( {"group": GROUP, "kind": kind, "limit": limit})) for m in (res or {}).get("memories", []): t = str(m.get("fact", "")).strip() if t: out.append(t) return out def pass_pattern(facts, feedback=""): """Pass 8 — pattern mining. ONE DeepSeek call over the timestamped corpus; saves up to 5 NEW kind='pattern' memories, each linked to 2-3 evidence ids, deduped against the patterns already stored (passed in as 'already known').""" known = _existing_kind("pattern") listing = _ts_listing(facts) if not listing: return 0 known_txt = "\n".join(f"- {t}" for t in known) or "(none yet)" raw = deepseek(feedback + PATTERN_PROMPT.format( known=known_txt, facts=listing), max_tokens=4000) # V4 reasoning room valid = {str(f.get("id")) for f in facts if f.get("id") is not None} created = 0 for it in _json_array(raw): if created >= 5: break if not isinstance(it, dict): continue text = str(it.get("text", "")).strip() if len(text) < 10 or _dup_task(text, known): continue ev = [str(x) for x in (it.get("evidence") or []) if str(x) in valid][:3] links = [{"to": e, "type": "evidence"} for e in ev] if DRY: print(f" [pattern] {text} <- {', '.join(ev) or 'no evidence'}", flush=True) known.append(text) created += 1 continue body = {"text": text, "kind": "pattern", "group": GROUP} if links: body["links"] = links if mem_opt("POST", "/remember", body) is not None: known.append(text) created += 1 return created def pass_synthesis(facts, feedback=""): """Pass 9 — reflection-tree. ONE DeepSeek call inferring 3-5 higher-level conclusions NOT directly stated, saved as kind='synthesis' linked to their sources and deduped against the syntheses already stored. Cap 5.""" known = _existing_kind("synthesis") listing = _id_listing(facts, cap=200) if not listing: return 0 known_txt = "\n".join(f"- {t}" for t in known) or "(none yet)" raw = deepseek(feedback + SYNTHESIS_PROMPT.format( known=known_txt, facts=listing), max_tokens=4000) # V4 reasoning room valid = {str(f.get("id")) for f in facts if f.get("id") is not None} created = 0 for it in _json_array(raw): if created >= 5: break if not isinstance(it, dict): continue text = str(it.get("text", "")).strip() if len(text) < 10 or _dup_task(text, known): continue src = [str(x) for x in (it.get("sources") or []) if str(x) in valid][:3] links = [{"to": s, "type": "source"} for s in src] if DRY: print(f" [synthesis] {text} <- {', '.join(src) or 'no src'}", flush=True) known.append(text) created += 1 continue body = {"text": text, "kind": "synthesis", "group": GROUP} if links: body["links"] = links if mem_opt("POST", "/remember", body) is not None: known.append(text) created += 1 return created def pass_core(facts, feedback=""): """Pass 10 — refresh the pinned CORE PROFILE (<=200 words) that gets injected into every Jarvis prompt. Built from the highest-value memories + tonight's fresh entity summaries; runs LAST so it reflects the night's cleanup. ONE DeepSeek call → POST /core. Returns True if a profile was written/printed.""" # highest-value memories (importance/access/recency) + the most recent. ranked = sorted(facts, key=_mem_rank, reverse=True)[:40] seen, top = set(), [] for f in ranked + facts[:20]: # facts are newest-first fid = str(f.get("id")) if fid in seen: continue seen.add(fid) txt = str(f.get("fact", "")).strip() if txt: top.append(f"- ({f.get('kind') or 'fact'}) {txt}") if len(top) >= 55: break facts_block = "\n".join(top) # tonight's summaries for the most-connected entities. graph = mem("GET", "/graph?" + urllib.parse.urlencode({"group": GROUP})) ents = {} for n in graph.get("nodes", []): nid = str(n.get("id", "")) if nid.startswith("ent:"): ents[nid] = {"name": str(n.get("text", "")).strip(), "deg": 0} for e in graph.get("edges", []): if e.get("type") == "mentions" and e.get("to") in ents: ents[e["to"]]["deg"] += 1 top_ents = [e for e in sorted(ents.values(), key=lambda x: -x["deg"]) if e["deg"] >= 2 and e["name"]][:12] sum_lines = [] for e in top_ents: try: r = mem("GET", "/entity?" + urllib.parse.urlencode( {"name": e["name"], "group": GROUP})) except Exception: # noqa: BLE001 continue for ent in (r or {}).get("entities", []): s = str(ent.get("summary", "")).strip() if s: sum_lines.append(f"- {e['name']}: {s}") break summaries_block = "\n".join(sum_lines) or "(none)" # prior core so the model refreshes (and drops stale) rather than reinvents. prior = "" cur = mem_opt("GET", "/core_block?" + urllib.parse.urlencode({"group": GROUP})) prior_txt = str((cur or {}).get("text", "")).strip() if cur else "" if prior_txt: prior = ("PRIOR CORE (refresh it — keep what still holds, DROP anything " "now stale):\n" + prior_txt + "\n\n") text = _clean_text(deepseek( feedback + CORE_PROMPT.format(prior=prior, summaries=summaries_block, facts=facts_block), max_tokens=2500)) if len(text) < 20: return False words = text.split() if len(words) > 230: # hard guard on the ~200-word budget text = " ".join(words[:230]) if DRY: print(f" [core] {text}", flush=True) return True return mem_opt("POST", "/core", {"text": text, "group": GROUP}) is not None def pass_archive(facts, graph, feedback=""): """Pass 11 — archive sweep. Retire memories that are ALL of: kind observation|event, importance<=4 (or absent), access_count 0 (or absent), created >90d ago, and NOT linked to another memory. Before retiring a CLUSTER (>=3 sharing an entity), one summary fact (kind='fact') is saved first so the gist survives. Cap 30 retirements/night. At most ONE DeepSeek call (only if there are clusters). Returns the retired count.""" cutoff = _dt.date.today() - _dt.timedelta(days=90) # memory-to-memory LINK ids (= "linked to something") + memory→entity mentions. linked, ment = set(), {} for e in graph.get("edges", []): fr, to, typ = str(e.get("from", "")), str(e.get("to", "")), e.get("type") if typ == "mentions": if not fr.startswith("ent:"): ment.setdefault(fr, set()).add(to) else: if not fr.startswith("ent:"): linked.add(fr) if not to.startswith("ent:"): linked.add(to) victims = [] for f in facts: fid = f.get("id") if fid is None: continue if str(f.get("kind") or "").lower() not in ("observation", "event"): continue imp = f.get("importance") if isinstance(imp, (int, float)) and imp > 4: continue ac = f.get("access_count") if isinstance(ac, (int, float)) and ac > 0: continue d = _parse_date(f.get("created_at")) if d is None or d >= cutoff: continue if str(fid) in linked: continue victims.append(f) if not victims: return 0 victims = victims[:30] # nightly cap vids = {str(f.get("id")) for f in victims} by_id = {str(f.get("id")): f for f in victims} # cluster victims that share an entity (>=3) → summarize before retiring. ent_groups = {} for vid in vids: for ent in ment.get(vid, ()): ent_groups.setdefault(ent, []).append(vid) clusters, clustered = [], set() for ent, ids in ent_groups.items(): ids = [i for i in ids if i not in clustered] if len(ids) >= 3: clusters.append((ent, ids)) clustered.update(ids) if clusters: blocks = [] for i, (ent, ids) in enumerate(clusters, 1): body = "; ".join(str(by_id[x].get("fact", "")).strip() for x in ids) blocks.append(f"Cluster {i} ({ent}): {body}") raw = deepseek(feedback + ARCHIVE_SUMMARY_PROMPT.format( clusters="\n".join(blocks)), max_tokens=2500) # V4 reasoning room summaries = {} for s in _json_array(raw): if not isinstance(s, dict): continue try: summaries[int(s.get("cluster", 0))] = str(s.get("text", "")).strip() except (TypeError, ValueError): continue for i, (ent, ids) in enumerate(clusters, 1): summ = summaries.get(i, "") if len(summ) < 10: continue links = [{"to": x, "type": "summary_of"} for x in ids] if DRY: print(f" [archive-summary] {summ} <- {', '.join(ids)}", flush=True) continue mem_opt("POST", "/remember", {"text": summ, "kind": "fact", "group": GROUP, "links": links}) retired = 0 for f in victims: if DRY: print(f" [archive] retire {f.get('id')}: " f"{str(f.get('fact', ''))[:60]}", flush=True) retired += 1 continue if mem_opt("POST", "/retire", {"id": f.get("id"), "group": GROUP}) is not None: retired += 1 return retired def _run(): facts_txt, ents_txt, facts, graph = build_context() n_facts = len(facts) if n_facts == 0: print("reflection: no memories yet", flush=True) return "no memories yet" # Ahmed's stated preferences — computed once, threaded into every generative # pass so the whole reflection respects what he's told Jarvis to do/stop. try: feedback = feedback_block() except Exception: # noqa: BLE001 feedback = "" # Pass 1 — insights (unchanged logic; prints moved to the combined summary). raw = deepseek(feedback + PROMPT.format(ents=ents_txt, facts=facts_txt)) m = re.search(r"\[.*\]", raw, re.DOTALL) try: findings = json.loads(m.group(0)) if m else [] except Exception: # noqa: BLE001 findings = [] findings = [f for f in findings if isinstance(f, dict)] created = 0 for f in findings: title = str(f.get("title", "")).strip() detail = str(f.get("detail", "")).strip() itype = str(f.get("type", "observation")).strip().lower() if len(detail) < 10: continue text = f"{title} — {detail}" if title else detail if DRY: print(f" [{itype}] {text}", flush=True) created += 1 continue r = mem("POST", "/insight", {"text": text, "title": title, "itype": itype, "group": GROUP}) if r.get("created"): created += 1 n_insights = created # Pass 2 — entity / community summaries. n_summaries = 0 try: n_summaries = pass_summaries(feedback) except Exception as e: # noqa: BLE001 print(f" [summary] pass failed: {e}", flush=True) # Pass 3 — goal / commitment extraction → tasks. n_goals = 0 try: n_goals = pass_goals(facts_txt, feedback) except Exception as e: # noqa: BLE001 print(f" [goal] pass failed: {e}", flush=True) # Pass 4 — consolidation (memory evolution / dedup). n_retired = 0 try: n_retired = pass_consolidate(facts) except Exception as e: # noqa: BLE001 print(f" [consolidate] pass failed: {e}", flush=True) # Pass 5 — theory-of-mind reads on the top people / companies. n_reads = 0 try: n_reads = pass_reads(feedback) except Exception as e: # noqa: BLE001 print(f" [read] pass failed: {e}", flush=True) # Pass 6 — Ahmed's north-star value model. values_ok = False try: values_ok = pass_values(facts_txt, feedback) except Exception as e: # noqa: BLE001 print(f" [values] pass failed: {e}", flush=True) # Pass 7 — entity hygiene (auto-dedup + adjudicate ambiguous band + retype flag). n_merges, retype_note = 0, "" try: n_merges, retype_note = pass_entity_hygiene(feedback) except Exception as e: # noqa: BLE001 print(f" [hygiene] pass failed: {e}", flush=True) # Pass 8 — pattern mining (recurring routines → kind=pattern). n_patterns = 0 try: n_patterns = pass_pattern(facts, feedback) except Exception as e: # noqa: BLE001 print(f" [pattern] pass failed: {e}", flush=True) # Pass 9 — synthesis (higher-level conclusions → kind=synthesis). n_synth = 0 try: n_synth = pass_synthesis(facts, feedback) except Exception as e: # noqa: BLE001 print(f" [synthesis] pass failed: {e}", flush=True) # Pass 10 — core block LAST, so it reflects tonight's dedup/merge/pattern work. core_ok = False try: core_ok = pass_core(facts, feedback) except Exception as e: # noqa: BLE001 print(f" [core] pass failed: {e}", flush=True) # Pass 11 — archive sweep (retire stale, low-value, isolated observations). n_archived = 0 try: n_archived = pass_archive(facts, graph, feedback) except Exception as e: # noqa: BLE001 print(f" [archive] pass failed: {e}", flush=True) note = (f"insights {n_insights}, summaries {n_summaries}, reads {n_reads}, " f"values {'y' if values_ok else 'n'}, goals {n_goals}, " f"consolidated {n_retired}, merges {n_merges}, patterns {n_patterns}, " f"synthesis {n_synth}, core {'y' if core_ok else 'n'}, " f"archived {n_archived}" + (f"; {retype_note}" if retype_note else "") + (" (dry)" if DRY else "")) print("reflection: " + note, flush=True) return note def main(): """Bracket the whole run in a heartbeat so a silent death is visible. started → (all passes) → ok; on ANY fatal exception, stamp error then re-raise. heartbeat() itself never raises, so the guard can't hide the real error.""" heartbeat("started") try: note = _run() except Exception as e: # noqa: BLE001 — stamp the failure, then re-raise heartbeat("error", f"{type(e).__name__}: {e}") raise heartbeat("ok", note) if __name__ == "__main__": main()