diff --git a/src/genesis/server/store.py b/src/genesis/server/store.py index dc56870..18a3ab9 100644 --- a/src/genesis/server/store.py +++ b/src/genesis/server/store.py @@ -150,13 +150,31 @@ class SessionStore: raise SessionNotFoundError(f"会话不存在: {session_id}") return self._from_dict(json.loads(row["data"])) - def list_sessions(self, user_id: str) -> list[SessionRecord]: - with self._conn() as c: - rows = c.execute( - "SELECT data FROM sessions WHERE user_id = ?", - (user_id,), - ).fetchall() - recs = [self._from_dict(json.loads(r["data"])) for r in rows] + def list_sessions(self, user_id: str, project: str | None = None) -> list[SessionRecord]: + if project is None: + with self._conn() as c: + rows = c.execute( + "SELECT data FROM sessions WHERE user_id = ?", + (user_id,), + ).fetchall() + recs = [self._from_dict(json.loads(r["data"])) for r in rows] + else: + try: + with self._conn() as c: + rows = c.execute( + "SELECT data FROM sessions WHERE user_id = ?" + " AND json_extract(data, '$.project') = ?", + (user_id, project), + ).fetchall() + recs = [self._from_dict(json.loads(r["data"])) for r in rows] + except Exception: + # json_extract 不可用(旧 SQLite)→ 全量取回后 Python 过滤兜底 + with self._conn() as c: + rows = c.execute( + "SELECT data FROM sessions WHERE user_id = ?", (user_id,) + ).fetchall() + recs = [self._from_dict(json.loads(r["data"])) for r in rows] + recs = [r for r in recs if r.project == project] # updated_at 在 JSON data 内,无法用 SQL 列排序 → 取回后按时间降序 recs.sort(key=lambda r: r.updated_at, reverse=True) return recs diff --git a/tests/test_server_store.py b/tests/test_server_store.py index aef09c2..e5452c3 100644 --- a/tests/test_server_store.py +++ b/tests/test_server_store.py @@ -213,3 +213,24 @@ def test_projects_upsert_empty_name_rejected(projects_store, tmp_path): with pytest.raises(ProjectConfigError): projects_store.upsert(name="", display_name="P", template="", write_instruction="", rules=[], existing_system_code_dir="", design_docs_dir="") + +def test_list_sessions_filter_by_project(store): + a = store.create_session("u1", name="A", project="stock") + b = store.create_session("u1", name="B", project="other") + recs = store.list_sessions("u1", project="stock") + ids = {r.session_id for r in recs} + assert a.session_id in ids and b.session_id not in ids + + +def test_list_sessions_filter_empty_project(store): + e = store.create_session("u1", name="E") + store.create_session("u1", name="X", project="stock") + recs = store.list_sessions("u1", project="") + assert all(r.project == "" for r in recs) + assert e.session_id in {r.session_id for r in recs} + + +def test_list_sessions_no_project_returns_all(store): + store.create_session("u1", name="A", project="stock") + store.create_session("u1", name="B", project="other") + assert len(store.list_sessions("u1")) == 2