Spaces:
Runtime error
Runtime error
File size: 4,862 Bytes
b74c6ae | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """The writer's standing notes ("audience: indie founders"), one JSON row per workspace.
Every chat loads them as background for tone and focus. They are never evidence: only posts are, and
citations still get checked. Notes come from the writer (Settings) or from the memory job, which reads
the writer's own messages and decides what is worth keeping. Notes the writer typed are never changed
by the job."""
import re
import uuid
from datetime import UTC, datetime
from fastapi import HTTPException
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.models import WorkspaceMemory
MAX_FACTS = 20
MAX_KEY = 60
MAX_VALUE = 500
PROFILE_CHARS = 1500
def normalize_key(key: str) -> str:
return re.sub(r"[^a-z0-9]+", "-", key.strip().lower()).strip("-")[:MAX_KEY]
def _now() -> str:
return datetime.now(UTC).isoformat()
def _row(db: Session, workspace_id: uuid.UUID, *, lock: bool = False, create: bool = False) -> WorkspaceMemory | None:
"""The workspace's row. lock=True holds it (FOR UPDATE on Postgres) so two writers cannot overwrite each other."""
q = select(WorkspaceMemory).where(WorkspaceMemory.workspace_id == workspace_id)
row = db.scalar(q.with_for_update() if lock else q)
if row or not create:
return row
db.add(WorkspaceMemory(workspace_id=workspace_id, facts={}))
try:
db.flush()
except IntegrityError: # another request created it first
db.rollback()
return db.scalar(q.with_for_update() if lock else q)
def _item(key: str, fact: dict) -> dict:
return {"key": key, "value": fact.get("value", ""), "source": fact.get("source", "user"),
"updated_at": fact.get("updated_at")}
def list_facts(db: Session, workspace_id: uuid.UUID) -> list[dict]:
row = _row(db, workspace_id)
items = [_item(k, f) for k, f in (row.facts or {}).items()] if row else []
return sorted(items, key=lambda i: i["updated_at"] or "", reverse=True)
def set_fact(db: Session, workspace_id: uuid.UUID, key: str, value: str, source: str = "user") -> dict:
"""Create or replace one note (the writer typing in Settings). Raises 422 on bad input."""
key, value = normalize_key(key), value.strip()
if not key:
raise HTTPException(422, "Give the note a short name")
if not value:
raise HTTPException(422, "The note cannot be empty")
if len(value) > MAX_VALUE:
raise HTTPException(422, f"Keep each note under {MAX_VALUE} characters")
row = _row(db, workspace_id, lock=True, create=True)
facts = dict(row.facts or {})
if key not in facts and len(facts) >= MAX_FACTS:
raise HTTPException(422, f"You can keep up to {MAX_FACTS} notes. Delete one to add another.")
facts[key] = {"value": value, "source": source, "updated_at": _now()}
row.facts = facts # a new dict, so the JSON column is saved
db.commit()
return _item(key, facts[key])
def delete_fact(db: Session, workspace_id: uuid.UUID, key: str) -> bool:
row = _row(db, workspace_id, lock=True)
key = normalize_key(key)
if not row or key not in (row.facts or {}):
return False
row.facts = {k: v for k, v in row.facts.items() if k != key}
db.commit()
return True
def apply_operations(db: Session, workspace_id: uuid.UUID, ops: list[dict]) -> list[dict]:
"""Apply the memory job's add/update/delete decisions. Bad or unsafe ones are skipped, never raised:
notes the writer typed stay as they are, the cap holds, and repeats change nothing. Returns what changed."""
row = _row(db, workspace_id, lock=True, create=True)
facts = dict(row.facts or {})
applied: list[dict] = []
for op in ops:
kind, key = op.get("op"), normalize_key(str(op.get("key") or ""))
value = str(op.get("value") or "").strip()
current = facts.get(key)
if not key or (current and current.get("source") == "user"):
continue
if kind in ("add", "update"):
if not value or len(value) > MAX_VALUE or (current and current.get("value") == value):
continue
if current is None and len(facts) >= MAX_FACTS:
continue
facts[key] = {"value": value, "source": "auto", "updated_at": _now()}
applied.append({"op": "update" if current else "add", "key": key, "value": value})
elif kind == "delete" and current:
del facts[key]
applied.append({"op": "delete", "key": key})
if applied:
row.facts = facts
db.commit()
return applied
def profile_text(items: list[dict]) -> str:
"""Prompt block for the agents: one line per note, capped so the prompt stays small."""
if not items:
return "(none)"
return "\n".join(f"- {i['key']}: {i['value']}" for i in items)[:PROFILE_CHARS]
|