host
cbc90f9d30
开关默认关（generate/use），设置接口 GET/PUT /users/me/auto-memory-settings，任务级覆盖 POST /tasks/{id}/auto-memory-flags，手动触发 POST /users/me/auto-memory/run，日志 GET /users/me/auto-memory-log。
提取：worker run_loop 轮次边界扫 completed 任务，闲置/时长/长度/外部上下文/配额剩余百分比五道门，脱敏后写 project 级 identity_memories，同一任务去重、同一目标 supersede 合并；只写 project 级，不跨项目泄漏。
注入：worker run_one 首轮把最新各 5 条用户/项目记忆拼进内存 prompt 副本（goal 任务同步改内存 goal），库原 prompt 不动；续跑/断点/协调员不注入，注入失败 fail-open。
v1 诚实边界：extract/consolidation 模型只存设置不调模型；MCP/web 线程内调用暂无 tasks 表标记，外部上下文门只认任务间消息与协调员子任务；无控制台 UI。
551 lines
23 KiB
Python
551 lines
23 KiB
Python
"""自动记忆：官方 Codex memories 的云端对齐版（v1）。
|
||
|
||
官方行为（learn.chatgpt.com/docs/customization/memories）：
|
||
- 开启后，Codex 把先前聊天里的有用上下文转成记忆，供未来工作用；
|
||
- 只对符合条件的先前聊天做，后台生成，不在每轮结束时立刻做；
|
||
- 聊天闲置够久才生成，避免总结仍在进行的工作；
|
||
- 剩余额度低于阈值时跳过本次生成，避免在额度见底时烧钱；
|
||
- 记忆存本地文件（~/.codex/memories/），有 generate / use /
|
||
disable_on_external_context / min_rate_limit_remaining_percent /
|
||
extract_model / consolidation_model 等设置；
|
||
- 密钥要脱敏，AGENTS.md 这类硬性规则不靠记忆。
|
||
|
||
云端对齐方式（本模块只做控制面数据库层，不动 host.py、
|
||
不改 CODEX_HOME、不改任何 thread 的创建/续跑/fork/归档）：
|
||
- generate_memories：完成任务是否可被提取成记忆。默认关，
|
||
与官方“本地记忆默认关”一致；用户在
|
||
PUT /users/me/auto-memory-settings 里显式打开才生效。
|
||
- use_memories：新任务首轮是否注入记忆。注入只改 worker 内存
|
||
里的任务副本（prompt 前加一段记忆前缀），库里的原 prompt
|
||
不动；续跑、断点续跑、协调员任务不注入（它们已有会话或
|
||
本身不跑模型回合）。
|
||
- disable_on_external_context：默认开。v1 能可靠识别的“外部
|
||
上下文”只有两类：任务间消息（task_followups 里有该任务的
|
||
行）与协调员子任务（parent_task_id 非空）。host 线程内的
|
||
MCP / web search 调用目前不在 tasks 表上留标记，本模块
|
||
诚实地不宣称能识别它们，见 docs/auto-memory.md 的未做。
|
||
- min_rate_limit_remaining_percent：官方的剩余额度门，云端
|
||
用当月 USD 配额剩余百分比对齐（与 worker 配额口径一致：
|
||
无配额行按默认 $10 计，只有显式 NULL 行才是不限额）。
|
||
提取本身是确定性文本处理、不调模型，但被提取的记忆随后
|
||
会被注入并占用后续任务的 prompt token，所以额度门仍在
|
||
提取处执行；查不到配额时 fail-closed（跳过并记
|
||
skipped_quota_check），不冒充“额度充足”。
|
||
- idle_seconds：任务终态后闲置多久才可提取，默认 900 秒。
|
||
终态后仍可被 /continue 续跑，所以“刚完成”不等于“已闲置”。
|
||
- 提取内容（v1 确定性版，不调模型）：从任务 goal/prompt 首行
|
||
与 result_text 前段拼一条 project 级 fact，脱敏后写入
|
||
identity_memories。extract_model / consolidation_model 只
|
||
存设置、不调模型——调模型提取是下一步显式决策（模型选择
|
||
与成本上限待定），v1 不假装已经调了。
|
||
- 去重与合并（consolidation 的确定性版）：同一任务只提取一次
|
||
（memory_extraction_log + 记忆行的 source_thread_id 记任务
|
||
id）；同一目标的新结果出现时，supersede 旧记忆而不是再
|
||
堆一条。自动提取只写 project 级记忆：单个项目的任务结论
|
||
若自动写成 user 级，会跨项目泄漏项目上下文，所以 user 级
|
||
记忆仍只接受手动创建。
|
||
- 任务级覆盖：POST /tasks/{id}/auto-memory-flags 可为单个任务
|
||
单独开关 use/generate（NULL = 跟随用户设置），不改 tasks
|
||
表结构（建任务的 4 条 INSERT 路径都不用动）。
|
||
|
||
新表一律 TEXT 主键，不走 IDENTITY 序列，见
|
||
cloud/migrations.py 的对齐说明；PG DDL 同步进
|
||
cloud/db.py POSTGRES_SCHEMA 与 MIGRATION_TABLES。
|
||
"""
|
||
|
||
import os
|
||
import re
|
||
import time
|
||
|
||
from cloud.db import connect as _db_connect
|
||
from cloud.identity import IdentityStore
|
||
|
||
DEFAULT_SETTINGS = {
|
||
"generate_memories": 0,
|
||
"use_memories": 0,
|
||
"disable_on_external_context": 1,
|
||
"min_rate_limit_remaining_percent": 20,
|
||
"idle_seconds": 900,
|
||
"min_duration_seconds": 30,
|
||
"min_result_chars": 200,
|
||
"extract_model": "",
|
||
"consolidation_model": "",
|
||
}
|
||
|
||
_BOOL_FIELDS = ("generate_memories", "use_memories",
|
||
"disable_on_external_context")
|
||
_INT_FIELDS = {
|
||
"min_rate_limit_remaining_percent": (0, 100),
|
||
"idle_seconds": (0, 86400),
|
||
"min_duration_seconds": (0, 86400),
|
||
"min_result_chars": (0, 10000),
|
||
}
|
||
_STR_FIELDS = ("extract_model", "consolidation_model")
|
||
|
||
# 脱敏：只处理“看起来像凭证”的形态，不把普通长 token（如
|
||
# git sha）一律抹掉。key=value 形态、Authorization/Bearer 头、
|
||
# 常见凭证前缀（sk- / ghp_ / gho_ / xoxb- / xoxp-）进黑名单。
|
||
_REDACT_PATTERNS = [
|
||
(re.compile(
|
||
r"(?i)(api[_-]?key|password|passwd|secret|token|credential)"
|
||
r"(\s*[:=]\s*)([\"']?)([^\s\"',;]+)"),
|
||
r"\1\2\3[REDACTED]"),
|
||
(re.compile(r"(?i)(authorization\s*:\s*)(Bearer\s+)?[^\s]+"),
|
||
r"\1[REDACTED]"),
|
||
(re.compile(r"(?i)\bBearer\s+[A-Za-z0-9._\-]+"),
|
||
"Bearer [REDACTED]"),
|
||
(re.compile(r"\b(sk-[A-Za-z0-9_\-]{8,}|ghp_[A-Za-z0-9]{8,}|"
|
||
r"gho_[A-Za-z0-9]{8,}|xox[baprs]-[A-Za-z0-9\-]{8,})"
|
||
r"\b"),
|
||
"[REDACTED]"),
|
||
]
|
||
|
||
|
||
def redact_text(text):
|
||
"""脱敏并清控制字符；非字符串返回空串。"""
|
||
if not isinstance(text, str):
|
||
return ""
|
||
out = text
|
||
for pattern, repl in _REDACT_PATTERNS:
|
||
out = pattern.sub(repl, out)
|
||
out = "".join(c for c in out if ord(c) >= 32 or c in ("\n", "\t"))
|
||
return out.strip()
|
||
|
||
|
||
def _default_quota_usd_micros():
|
||
"""与 worker/server 同款默认配额解析，避免循环 import。"""
|
||
try:
|
||
return max(0, int(os.environ.get(
|
||
"CLOUD_DEFAULT_QUOTA_USD_MICROS", "10000000")))
|
||
except (TypeError, ValueError):
|
||
return 10000000
|
||
|
||
|
||
def quota_remaining_percent(db_path, owner):
|
||
"""返回当月配额剩余百分比；不限额返回 None。
|
||
|
||
查库失败抛异常（调用方 fail-closed），全新库无成本行时
|
||
已用按 0 计（那是真没有花费，不是查失败）。
|
||
"""
|
||
import datetime
|
||
month = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m")
|
||
conn = _db_connect(db_path)
|
||
try:
|
||
conn.execute(
|
||
"CREATE TABLE IF NOT EXISTS user_quotas("
|
||
"user_id TEXT PRIMARY KEY, quota_usd_micros BIGINT, "
|
||
"updated_at DOUBLE PRECISION)")
|
||
conn.commit()
|
||
row = conn.execute(
|
||
"SELECT quota_usd_micros FROM user_quotas WHERE user_id = ?",
|
||
(owner,)).fetchone()
|
||
quota = _default_quota_usd_micros() if row is None else row[0]
|
||
if quota is None:
|
||
return None
|
||
if quota <= 0:
|
||
return 0.0
|
||
try:
|
||
used_row = conn.execute(
|
||
"SELECT COALESCE(SUM(c.usd_micros), 0)"
|
||
" FROM cost_records c JOIN tasks t"
|
||
" ON c.task_id = t.id"
|
||
" WHERE t.created_by = ? AND c.day LIKE ?",
|
||
(owner, month + "%")).fetchone()
|
||
used = used_row[0] if used_row else 0
|
||
except Exception as exc:
|
||
if "no such table" in str(exc):
|
||
used = 0
|
||
else:
|
||
raise
|
||
return max(0.0, min(100.0, (quota - int(used or 0)) * 100.0 / quota))
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _row_get(row, name, default=None):
|
||
try:
|
||
return row[name]
|
||
except (KeyError, IndexError, TypeError):
|
||
return default
|
||
|
||
|
||
def _first_line(text, max_len=200):
|
||
for line in (text or "").splitlines():
|
||
line = " ".join(line.split())
|
||
if line:
|
||
return redact_text(line)[:max_len]
|
||
return ""
|
||
|
||
|
||
def build_extraction_content(task, result_text):
|
||
"""从任务拼一条可存的记忆正文；拼不出返回空串。"""
|
||
goal = _first_line(task.get("goal") or task.get("prompt") or "")
|
||
if not goal:
|
||
return ""
|
||
outcome = " ".join((result_text or "").split())
|
||
outcome = redact_text(outcome)[:800]
|
||
if outcome:
|
||
content = "任务完成：%s；结果：%s" % (goal, outcome)
|
||
else:
|
||
content = "任务完成：%s" % goal
|
||
return content[:3900]
|
||
|
||
|
||
class AutoMemoryStore:
|
||
"""记忆设置、任务级覆盖、提取日志（SQLite 或 Postgres）。"""
|
||
|
||
def __init__(self, db_path):
|
||
self.db_path = db_path
|
||
self.conn = _db_connect(db_path)
|
||
self.conn.execute(
|
||
"CREATE TABLE IF NOT EXISTS memory_settings("
|
||
"user_id TEXT PRIMARY KEY, "
|
||
"generate_memories INTEGER DEFAULT 0, "
|
||
"use_memories INTEGER DEFAULT 0, "
|
||
"disable_on_external_context INTEGER DEFAULT 1, "
|
||
"min_rate_limit_remaining_percent INTEGER DEFAULT 20, "
|
||
"idle_seconds INTEGER DEFAULT 900, "
|
||
"min_duration_seconds REAL DEFAULT 30, "
|
||
"min_result_chars INTEGER DEFAULT 200, "
|
||
"extract_model TEXT, "
|
||
"consolidation_model TEXT, "
|
||
"updated_at REAL)"
|
||
)
|
||
self.conn.execute(
|
||
"CREATE TABLE IF NOT EXISTS task_memory_flags("
|
||
"task_id TEXT PRIMARY KEY, "
|
||
"use_memories INTEGER, "
|
||
"generate_memories INTEGER, "
|
||
"updated_at REAL)"
|
||
)
|
||
self.conn.execute(
|
||
"CREATE TABLE IF NOT EXISTS memory_extraction_log("
|
||
"task_id TEXT PRIMARY KEY, "
|
||
"owner_user_id TEXT, "
|
||
"project_id TEXT, "
|
||
"status TEXT, "
|
||
"reason TEXT, "
|
||
"memory_id TEXT, "
|
||
"created_at REAL, "
|
||
"processed_at REAL)"
|
||
)
|
||
self.conn.commit()
|
||
|
||
def close(self):
|
||
try:
|
||
self.conn.close()
|
||
except Exception:
|
||
pass
|
||
|
||
def get_settings(self, user_id):
|
||
row = self.conn.execute(
|
||
"SELECT * FROM memory_settings WHERE user_id = ?",
|
||
(user_id,)).fetchone()
|
||
settings = dict(DEFAULT_SETTINGS)
|
||
if row is not None:
|
||
d = dict(row)
|
||
for key in DEFAULT_SETTINGS:
|
||
if d.get(key) is not None:
|
||
settings[key] = d[key]
|
||
settings["user_id"] = user_id
|
||
return settings
|
||
|
||
def upsert_settings(self, user_id, **kwargs):
|
||
"""只改给出的字段；非法值抛 ValueError，不落库。"""
|
||
clean = {}
|
||
for key, value in kwargs.items():
|
||
if key in _BOOL_FIELDS:
|
||
if value not in (True, False, 0, 1):
|
||
raise ValueError("%s must be a boolean" % key)
|
||
clean[key] = 1 if value else 0
|
||
elif key in _INT_FIELDS:
|
||
lo, hi = _INT_FIELDS[key]
|
||
try:
|
||
ivalue = int(value)
|
||
except (TypeError, ValueError):
|
||
raise ValueError("%s must be an integer" % key)
|
||
if isinstance(value, bool) or not lo <= ivalue <= hi:
|
||
raise ValueError(
|
||
"%s must be between %d and %d" % (key, lo, hi))
|
||
clean[key] = ivalue
|
||
elif key in _STR_FIELDS:
|
||
if value is None:
|
||
clean[key] = ""
|
||
elif not isinstance(value, str) or len(value) > 128:
|
||
raise ValueError("%s must be a short string" % key)
|
||
else:
|
||
clean[key] = value.strip()
|
||
else:
|
||
raise ValueError("unknown setting: %s" % key)
|
||
existing = self.get_settings(user_id)
|
||
merged = dict(existing)
|
||
merged.update(clean)
|
||
now = time.time()
|
||
self.conn.execute(
|
||
"INSERT OR REPLACE INTO memory_settings("
|
||
"user_id, generate_memories, use_memories, "
|
||
"disable_on_external_context, "
|
||
"min_rate_limit_remaining_percent, idle_seconds, "
|
||
"min_duration_seconds, min_result_chars, extract_model, "
|
||
"consolidation_model, updated_at) "
|
||
"VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(user_id, merged["generate_memories"], merged["use_memories"],
|
||
merged["disable_on_external_context"],
|
||
merged["min_rate_limit_remaining_percent"],
|
||
merged["idle_seconds"], merged["min_duration_seconds"],
|
||
merged["min_result_chars"], merged["extract_model"],
|
||
merged["consolidation_model"], now))
|
||
self.conn.commit()
|
||
return self.get_settings(user_id)
|
||
|
||
def get_task_flags(self, task_id):
|
||
row = self.conn.execute(
|
||
"SELECT * FROM task_memory_flags WHERE task_id = ?",
|
||
(task_id,)).fetchone()
|
||
return dict(row) if row is not None else None
|
||
|
||
def set_task_flags(self, task_id, use_memories=None,
|
||
generate_memories=None):
|
||
if use_memories is None and generate_memories is None:
|
||
raise ValueError(
|
||
"at least one of use_memories / generate_memories required")
|
||
for name, value in (("use_memories", use_memories),
|
||
("generate_memories", generate_memories)):
|
||
if value is not None and value not in (True, False, 0, 1):
|
||
raise ValueError("%s must be a boolean" % name)
|
||
old = self.get_task_flags(task_id) or {}
|
||
self.conn.execute(
|
||
"INSERT OR REPLACE INTO task_memory_flags("
|
||
"task_id, use_memories, generate_memories, updated_at) "
|
||
"VALUES(?, ?, ?, ?)",
|
||
(task_id,
|
||
(1 if use_memories else 0) if use_memories is not None
|
||
else old.get("use_memories"),
|
||
(1 if generate_memories else 0)
|
||
if generate_memories is not None
|
||
else old.get("generate_memories"),
|
||
time.time()))
|
||
self.conn.commit()
|
||
return self.get_task_flags(task_id)
|
||
|
||
@staticmethod
|
||
def effective(settings, flags, field):
|
||
"""任务级覆盖优先于用户设置；NULL = 跟随设置。"""
|
||
if flags is not None and flags.get(field) is not None:
|
||
return bool(flags[field])
|
||
return bool(settings[field])
|
||
|
||
def write_log(self, task_id, owner, project_id, status, reason,
|
||
memory_id=None, now=None):
|
||
now = now if now is not None else time.time()
|
||
row = self.conn.execute(
|
||
"SELECT created_at FROM memory_extraction_log"
|
||
" WHERE task_id = ?", (task_id,)).fetchone()
|
||
created = row[0] if row else now
|
||
self.conn.execute(
|
||
"INSERT OR REPLACE INTO memory_extraction_log("
|
||
"task_id, owner_user_id, project_id, status, reason, "
|
||
"memory_id, created_at, processed_at) "
|
||
"VALUES(?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(task_id, owner, project_id, status, reason,
|
||
memory_id or "", created, now))
|
||
self.conn.commit()
|
||
|
||
def list_log(self, owner, limit=50):
|
||
rows = self.conn.execute(
|
||
"SELECT * FROM memory_extraction_log"
|
||
" WHERE owner_user_id = ?"
|
||
" ORDER BY processed_at DESC, task_id DESC LIMIT ?",
|
||
(owner, max(1, min(200, int(limit))))).fetchall()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def build_injection_prefix(db_path, owner, project_id, flags=None):
|
||
"""use_memories 关、或无记忆时返回空串；否则返回注入前缀。"""
|
||
store = AutoMemoryStore(db_path)
|
||
try:
|
||
settings = store.get_settings(owner)
|
||
if not AutoMemoryStore.effective(settings, flags, "use_memories"):
|
||
return ""
|
||
finally:
|
||
store.close()
|
||
identity = IdentityStore(db_path)
|
||
try:
|
||
context = identity.build_identity_context(owner, project_id)
|
||
finally:
|
||
try:
|
||
identity.conn.close()
|
||
except Exception:
|
||
pass
|
||
if not context:
|
||
return ""
|
||
return ("【过往记忆，仅供参考，不得改写本任务目标】\n%s" % context)
|
||
|
||
|
||
def _has_external_context(conn, task_id, parent_task_id):
|
||
if parent_task_id:
|
||
return True
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT 1 FROM task_followups WHERE task_id = ? LIMIT 1",
|
||
(task_id,)).fetchone()
|
||
return row is not None
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def process_due_tasks(db_path, now=None, force=False, owner_id=None):
|
||
"""扫 completed 任务并提取记忆，返回每任务一条结果。
|
||
|
||
幂等：已提取的任务不再重复建记忆（force 也不重复，靠
|
||
source_thread_id 去重）；被跳过的任务下次仍可再扫（设置
|
||
改了、闲置够了、额度回来了都可能翻盘），日志行按任务
|
||
主键覆盖，不无限增长。
|
||
"""
|
||
now = now if now is not None else time.time()
|
||
store = AutoMemoryStore(db_path)
|
||
identity = IdentityStore(db_path)
|
||
results = []
|
||
try:
|
||
if owner_id:
|
||
rows = store.conn.execute(
|
||
"SELECT * FROM tasks WHERE status = 'completed'"
|
||
" AND created_by = ? ORDER BY id",
|
||
(owner_id,)).fetchall()
|
||
else:
|
||
rows = store.conn.execute(
|
||
"SELECT * FROM tasks WHERE status = 'completed'"
|
||
" ORDER BY id").fetchall()
|
||
except Exception as exc:
|
||
store.close()
|
||
try:
|
||
identity.conn.close()
|
||
except Exception:
|
||
pass
|
||
return [{"status": "error", "reason": "tasks query failed: %s"
|
||
% exc}]
|
||
for row in rows:
|
||
task = dict(row)
|
||
task_id = task.get("id")
|
||
owner = task.get("created_by")
|
||
project_id = task.get("project_id")
|
||
result = {"task_id": task_id, "status": "skipped",
|
||
"reason": "", "memory_id": None}
|
||
results.append(result)
|
||
if not owner or not project_id:
|
||
result["reason"] = "no_owner_or_project"
|
||
store.write_log(task_id, owner or "", project_id or "",
|
||
"skipped", result["reason"], now=now)
|
||
continue
|
||
settings = store.get_settings(owner)
|
||
flags = store.get_task_flags(task_id)
|
||
if not AutoMemoryStore.effective(
|
||
settings, flags, "generate_memories"):
|
||
result["reason"] = "generate_disabled"
|
||
store.write_log(task_id, owner, project_id, "skipped",
|
||
result["reason"], now=now)
|
||
continue
|
||
finished = _row_get(task, "finished_at") \
|
||
or _row_get(task, "updated_at") \
|
||
or _row_get(task, "created_at")
|
||
started = _row_get(task, "started_at")
|
||
if not force:
|
||
if finished is None or now - float(finished) < \
|
||
float(settings["idle_seconds"]):
|
||
result["reason"] = "idle_not_reached"
|
||
store.write_log(task_id, owner, project_id, "skipped",
|
||
result["reason"], now=now)
|
||
continue
|
||
if started is None or finished is None:
|
||
result["reason"] = "no_timing"
|
||
store.write_log(task_id, owner, project_id, "skipped",
|
||
result["reason"], now=now)
|
||
continue
|
||
duration = float(finished) - float(started)
|
||
if duration < float(settings["min_duration_seconds"]):
|
||
result["reason"] = "too_short"
|
||
store.write_log(task_id, owner, project_id, "skipped",
|
||
result["reason"], now=now)
|
||
continue
|
||
result_text = _row_get(task, "result_text") or ""
|
||
if len(result_text) < int(settings["min_result_chars"]) \
|
||
and not force:
|
||
result["reason"] = "result_too_short"
|
||
store.write_log(task_id, owner, project_id, "skipped",
|
||
result["reason"], now=now)
|
||
continue
|
||
if settings["disable_on_external_context"] and \
|
||
_has_external_context(
|
||
store.conn, task_id,
|
||
_row_get(task, "parent_task_id")):
|
||
result["reason"] = "external_context"
|
||
store.write_log(task_id, owner, project_id, "skipped",
|
||
result["reason"], now=now)
|
||
continue
|
||
min_remaining = int(
|
||
settings["min_rate_limit_remaining_percent"])
|
||
if min_remaining > 0:
|
||
try:
|
||
remaining = quota_remaining_percent(db_path, owner)
|
||
except Exception:
|
||
result["reason"] = "quota_check_failed"
|
||
store.write_log(task_id, owner, project_id, "skipped",
|
||
result["reason"], now=now)
|
||
continue
|
||
if remaining is not None and remaining < min_remaining:
|
||
result["reason"] = "quota_low"
|
||
store.write_log(task_id, owner, project_id, "skipped",
|
||
result["reason"], now=now)
|
||
continue
|
||
content = build_extraction_content(task, result_text)
|
||
if not content:
|
||
result["reason"] = "empty_content"
|
||
store.write_log(task_id, owner, project_id, "skipped",
|
||
result["reason"], now=now)
|
||
continue
|
||
existing = identity.list_project_memories(project_id)
|
||
same_source = [r for r in existing
|
||
if r.get("source_thread_id") == task_id]
|
||
if same_source:
|
||
result["status"] = "extracted"
|
||
result["reason"] = "already_extracted"
|
||
result["memory_id"] = same_source[0]["id"]
|
||
store.write_log(task_id, owner, project_id, "extracted",
|
||
result["reason"], result["memory_id"], now=now)
|
||
continue
|
||
goal_line = _first_line(task.get("goal") or task.get("prompt")
|
||
or "")
|
||
same_goal = [r for r in existing
|
||
if r.get("content", "").startswith(
|
||
"任务完成：%s；" % goal_line)] if goal_line else []
|
||
if same_goal:
|
||
_old, new_id = identity.supersede_memory(
|
||
same_goal[0]["id"], owner,
|
||
replacement_content=content)
|
||
if new_id:
|
||
try:
|
||
identity.conn.execute(
|
||
"UPDATE identity_memories SET source_thread_id = ?"
|
||
" WHERE id = ?", (task_id, new_id))
|
||
identity.conn.commit()
|
||
except Exception:
|
||
pass
|
||
result["status"] = "extracted"
|
||
result["reason"] = "consolidated"
|
||
result["memory_id"] = new_id
|
||
else:
|
||
new_id = identity.add_memory(
|
||
owner, "project", "fact", content,
|
||
project_id=project_id, source_thread_id=task_id)
|
||
result["status"] = "extracted"
|
||
result["reason"] = "created"
|
||
result["memory_id"] = new_id
|
||
store.write_log(task_id, owner, project_id, "extracted",
|
||
result["reason"], result["memory_id"], now=now)
|
||
store.close()
|
||
try:
|
||
identity.conn.close()
|
||
except Exception:
|
||
pass
|
||
return results
|