Coverage for src \ truenex_memory \ cli \ task_commands.py: 17%

1"""CLI commands for the adaptive task pipeline (truenex-mem task ...).""" 

2from __future__ import annotations 

3from pathlib import Path 

4import json 

5import typer 

6from truenex_memory.store.task_store import TaskStore, TASK_TYPES 

7 

8task_app = typer.Typer(help="Manage adaptive task pipeline records.") 

9_DEFAULT_DB = Path.home() / ".truenex-memory" / "truenex_memory.db" 

10 

11 

12def _store(db: Path | None = None) -> TaskStore: 

13 return TaskStore(db or _DEFAULT_DB) 

14 

15 

16@task_app.command("open") 

17def task_open( 

18 title: str = typer.Argument(..., help="Short task description."), 

19 task_type: str = typer.Option("feature", "--type", "-t", help="Task type: bugfix|feature|refactor|review|query"), 

20 project: str = typer.Option(None, "--project", "-p", help="Project name."), 

21 session: str = typer.Option(None, "--session", help="Agent session ID."), 

22 json_out: bool = typer.Option(False, "--json", help="Output as JSON."), 

23) -> None: 

24 """Open a new task record.""" 

25 if task_type not in TASK_TYPES: 

26 typer.echo(f"Error: --type must be one of {sorted(TASK_TYPES)}", err=True) 

27 raise typer.Exit(code=1) 

28 task_id = _store().task_open(title, task_type, project=project, agent_session_id=session) 

29 if json_out: 

30 typer.echo(json.dumps({"task_id": task_id, "status": "open"}, indent=2)) 

31 else: 

32 typer.echo(f"Task opened: {task_id}") 

33 

34 

35@task_app.command("close") 

36def task_close( 

37 task_id: str = typer.Argument(..., help="Task ID to close."), 

38 outcome: int = typer.Option(None, "--outcome", "-o", help="Human outcome: 1=positive, 0=partial, -1=negative."), 

39 comment: str = typer.Option(None, "--comment", "-c", help="Optional human comment."), 

40 unrated: bool = typer.Option(False, "--unrated", help="Mark as unrated (no judgment)."), 

41 json_out: bool = typer.Option(False, "--json"), 

42) -> None: 

43 """Close a task with optional human judgment.""" 

44 store = _store() 

45 if unrated: 

46 human_outcome = None 

47 human_comment = None 

48 elif outcome is not None: 

49 if outcome not in (1, 0, -1): 

50 typer.echo("Error: --outcome must be 1, 0, or -1", err=True) 

51 raise typer.Exit(code=1) 

52 human_outcome = outcome 

53 human_comment = comment 

54 else: 

55 try: 

56 task = store.task_get(task_id) 

57 except LookupError as exc: 

58 typer.echo(f"Error: {exc}", err=True) 

59 raise typer.Exit(code=1) 

60 typer.echo(f"\nTask: {task.title}") 

61 typer.echo(f"Steps: {len(store.step_list(task_id))}") 

62 raw = typer.prompt("Human outcome (1=positive, 0=partial, -1=negative, skip=unrated)", default="skip") 

63 if raw.strip().lower() in ("skip", ""): 

64 human_outcome = None 

65 human_comment = None 

66 else: 

67 try: 

68 human_outcome = int(raw) 

69 if human_outcome not in (1, 0, -1): 

70 raise ValueError 

71 except ValueError: 

72 typer.echo("Invalid input — marking as unrated.", err=True) 

73 human_outcome = None 

74 human_comment = typer.prompt("Comment (optional, press Enter to skip)", default="") or None 

75 try: 

76 record = store.task_close(task_id, human_outcome=human_outcome, human_comment=human_comment) 

77 except LookupError as exc: 

78 typer.echo(f"Error: {exc}", err=True) 

79 raise typer.Exit(code=1) 

80 if json_out: 

81 typer.echo(json.dumps({"task_id": record.task_id, "status": record.status, "human_outcome": record.human_outcome}, indent=2)) 

82 else: 

83 typer.echo(f"Task closed: {record.task_id} — status={record.status}, outcome={record.human_outcome}") 

84 

85 

86@task_app.command("list") 

87def task_list( 

88 project: str = typer.Option(None, "--project", "-p"), 

89 status: str = typer.Option(None, "--status", "-s", help="Filter: open|closed|unrated"), 

90 limit: int = typer.Option(20, "--limit", "-n"), 

91 json_out: bool = typer.Option(False, "--json"), 

92) -> None: 

93 """List recent tasks.""" 

94 try: 

95 records = _store().task_list(project=project, status=status, limit=limit) 

96 except ValueError as exc: 

97 typer.echo(f"Error: {exc}", err=True) 

98 raise typer.Exit(code=1) 

99 if json_out: 

100 typer.echo(json.dumps([r.__dict__ for r in records], indent=2)) 

101 return 

102 if not records: 

103 typer.echo("No tasks found.") 

104 return 

105 for r in records: 

106 outcome = r.human_outcome if r.human_outcome is not None else "?" 

107 typer.echo(f"{r.task_id} [{r.status}] [{r.type}] outcome={outcome} {r.title[:60]}") 

108 

109 

110@task_app.command("show") 

111def task_show( 

112 task_id: str = typer.Argument(..., help="Task ID."), 

113 json_out: bool = typer.Option(False, "--json"), 

114) -> None: 

115 """Show task details with steps.""" 

116 store = _store() 

117 try: 

118 task = store.task_get(task_id) 

119 steps = store.step_list(task_id) 

120 except LookupError as exc: 

121 typer.echo(f"Error: {exc}", err=True) 

122 raise typer.Exit(code=1) 

123 if json_out: 

124 typer.echo(json.dumps({"task": task.__dict__, "steps": [s.__dict__ for s in steps]}, indent=2)) 

125 return 

126 typer.echo(f"Task: {task.task_id}") 

127 typer.echo(f"Title: {task.title}") 

128 typer.echo(f"Type: {task.type} Status: {task.status}") 

129 typer.echo(f"Project: {task.project or 'N/A'}") 

130 typer.echo(f"Outcome: {task.human_outcome} Comment: {task.human_comment or 'N/A'}") 

131 typer.echo(f"Tokens: {task.total_tokens} Duration: {task.total_duration_s}s") 

132 typer.echo(f"Created: {task.created_at} Closed: {task.closed_at or 'N/A'}") 

133 typer.echo(f"\nSteps ({len(steps)}):") 

134 for s in steps: 

135 typer.echo(f" [{s.step_index}] judgment={s.brain_judgment} tokens={s.tokens_used} model={s.model_used}") 

136 

137 

138@task_app.command("calibration") 

139def task_calibration( 

140 project: str = typer.Option(None, "--project", "-p"), 

141 json_out: bool = typer.Option(False, "--json"), 

142) -> None: 

143 """Show calibration stats.""" 

144 data = _store().calibration(project=project) 

145 if json_out: 

146 typer.echo(json.dumps(data, indent=2)) 

147 return 

148 typer.echo("=== Verifier Acceptance ===") 

149 for row in data["verifier_acceptance"]: 

150 rate = row["acceptance_rate"] 

151 rate_str = f"{rate:.1%}" if rate is not None else "N/A" 

152 typer.echo(f" {row['suggestion_type']}: {rate_str} ({row['accepted']}/{row['total']})") 

153 typer.echo("\n=== Brain vs Human Alignment ===") 

154 for row in data["brain_human_alignment"]: 

155 typer.echo(f" brain={row['brain_judgment']} human={row['human_outcome']}: {row['count']}")