Coverage for src \ truenex_memory \ mcp \ tools.py: 71%

1"""Local MCP-compatible tool functions.""" 

2 

3from __future__ import annotations 

4 

5from pathlib import Path 

6 

7from truenex_memory.core.memory_service import MemoryService 

8from truenex_memory.ingestion.global_context import build_project_context 

9from truenex_memory.ingestion.global_status import build_global_status 

10from truenex_memory.retrieval.result import search_payload 

11 

12 

13def memory_search(query: str, top_k: int = 5, *, project_root: Path | str = ".") -> dict[str, object]: 

14 """Search local memory using the stable MCP result shape.""" 

15 

16 service = MemoryService(project_root) 

17 results = service.search(query, top_k=top_k) 

18 return search_payload(query, results, trace_id=service.last_trace_id) 

19 

20 

21def memory_add( 

22 content: str, 

23 memory_type: str = "note", 

24 *, 

25 project_root: Path | str = ".", 

26) -> dict[str, object]: 

27 """Add a local memory node.""" 

28 

29 service = MemoryService(project_root) 

30 memory_id = service.add(content, memory_type=memory_type) 

31 return {"id": memory_id, "status": "active", "memory_type": memory_type} 

32 

33 

34def global_status( 

35 home: str | Path | None = None, 

36 catalog: str | Path | None = None, 

37 db: str | Path | None = None, 

38) -> dict[str, object]: 

39 """Read-only global status report for the Truenex Memory global store.""" 

40 

41 _home = Path(home) if home else Path.home() 

42 catalog_path = Path(catalog) if catalog else _home / ".truenex-memory" / "sources.json" 

43 db_path = Path(db) if db else _home / ".truenex-memory" / "truenex_memory.db" 

44 

45 report = build_global_status(catalog_path=catalog_path, db_path=db_path) 

46 return report.to_dict() 

47 

48 

49def global_project_context( 

50 project: str, 

51 home: str | Path | None = None, 

52 catalog: str | Path | None = None, 

53 db: str | Path | None = None, 

54 limit: int = 20, 

55) -> dict[str, object]: 

56 """Read-only project context report for a project in the global store.""" 

57 

58 if not isinstance(project, str) or not project.strip(): 

59 raise ValueError("project must be a non-empty string") 

60 if not isinstance(limit, int) or limit < 1 or limit > 100: 

61 raise ValueError("limit must be an integer between 1 and 100") 

62 

63 _home = Path(home) if home else Path.home() 

64 catalog_path = Path(catalog) if catalog else _home / ".truenex-memory" / "sources.json" 

65 db_path = Path(db) if db else _home / ".truenex-memory" / "truenex_memory.db" 

66 

67 report = build_project_context( 

68 project_query=project, 

69 catalog_path=catalog_path, 

70 db_path=db_path, 

71 limit=limit, 

72 ) 

73 return report.to_dict() 

74 

75 

76from truenex_memory.store.task_store import TaskStore, TASK_TYPES, BRAIN_JUDGMENTS 

77 

78 

79def _default_task_store(db: str | None = None) -> TaskStore: 

80 db_path = Path(db) if db else Path.home() / ".truenex-memory" / "truenex_memory.db" 

81 return TaskStore(db_path) 

82 

83 

84def task_open( 

85 title: str, 

86 task_type: str = "feature", 

87 *, 

88 project: str | None = None, 

89 agent_session_id: str | None = None, 

90 db: str | None = None, 

91) -> dict[str, object]: 

92 """Open a new task record in the adaptive pipeline.""" 

93 if task_type not in TASK_TYPES: 

94 raise ValueError(f"task_type must be one of {sorted(TASK_TYPES)}") 

95 task_id = _default_task_store(db).task_open(title, task_type, project=project, agent_session_id=agent_session_id) 

96 return {"task_id": task_id, "status": "open"} 

97 

98 

99def task_step_add( 

100 task_id: str, 

101 *, 

102 prompt_used: str | None = None, 

103 output: str | None = None, 

104 brain_judgment: str | None = None, 

105 tokens_used: int | None = None, 

106 duration_s: float | None = None, 

107 model_used: str | None = None, 

108 db: str | None = None, 

109) -> dict[str, object]: 

110 """Add a step record to an open task.""" 

111 if brain_judgment is not None and brain_judgment not in BRAIN_JUDGMENTS: 

112 raise ValueError(f"brain_judgment must be one of {sorted(BRAIN_JUDGMENTS)}") 

113 step_id = _default_task_store(db).step_add( 

114 task_id, prompt_used=prompt_used, output=output, brain_judgment=brain_judgment, 

115 tokens_used=tokens_used, duration_s=duration_s, model_used=model_used, 

116 ) 

117 return {"step_id": step_id, "task_id": task_id} 

118 

119 

120def task_close( 

121 task_id: str, 

122 *, 

123 human_outcome: int | None = None, 

124 human_comment: str | None = None, 

125 db: str | None = None, 

126) -> dict[str, object]: 

127 """Close a task. Provide human_outcome (1/0/-1) or omit for unrated.""" 

128 if human_outcome is not None and human_outcome not in (1, 0, -1): 

129 raise ValueError("human_outcome must be 1, 0, or -1") 

130 record = _default_task_store(db).task_close(task_id, human_outcome=human_outcome, human_comment=human_comment) 

131 return {"task_id": record.task_id, "status": record.status, "human_outcome": record.human_outcome, "closed_at": record.closed_at}