Skip to main content

max / goingson

20.7 KB · 603 lines History Blame Raw
1 //! Task tools: the read surface (`list_tasks`, `get_task`) and the write
2 //! surface (`create_task`, `bulk_import_tasks`, `update_task`, `complete_task`).
3 //!
4 //! `bulk_import_tasks` is the `/dellm` migration primitive: it takes a parsed
5 //! backlog and creates tasks in one capability-gated call, deduping on a
6 //! `source:` provenance tag so a re-run does not double-insert.
7
8 use std::collections::{HashMap, HashSet};
9 use std::sync::Arc;
10
11 use async_trait::async_trait;
12 use goingson_core::models::ParseableEnum;
13 use goingson_core::repository::{ProjectRepository, TaskAnnotations, TaskCrud};
14 use goingson_core::{
15 NewProject, NewTask, ProjectId, ProjectStatus, ProjectType, TOKEN_KIND_COMMIT, Task,
16 TaskStatus, TokenState, UpdateTask,
17 };
18 use kberg::{Error, Result, Tool, ToolCallResult, ToolKind};
19 use serde_json::{Value, json};
20
21 use crate::caps;
22 use crate::context::Ctx;
23 use crate::convert::{
24 MAX_LIMIT, parse_due, parse_limit, parse_offset, parse_priority, parse_tags, parse_task_id,
25 req_str, source_tag, task_row, task_summary_row,
26 };
27
28 fn fail(tool: &str, e: impl std::fmt::Display) -> Error {
29 Error::ToolFailed {
30 tool: tool.to_string(),
31 message: e.to_string(),
32 }
33 }
34
35 // reads
36
37 pub struct ListTasks(pub Arc<Ctx>);
38
39 #[async_trait]
40 impl Tool for ListTasks {
41 fn name(&self) -> &str {
42 "list_tasks"
43 }
44 fn description(&self) -> &str {
45 "List tasks as compact rows. Optional filters: `project` (name), `status` (Pending|Started|Completed), `tag`. Paged: `limit` (default 50, max 200) and `offset`. Long descriptions are clipped and the row marked `truncated`; use `get_task` for the full text. The reply carries `total` (rows matching the filters) and, when more remain, `next_offset`."
46 }
47 fn kind(&self) -> ToolKind {
48 ToolKind::Read
49 }
50 fn small_model_safe(&self) -> bool {
51 true
52 }
53 fn input_schema(&self) -> Value {
54 json!({
55 "type": "object",
56 "properties": {
57 "project": { "type": "string" },
58 "status": { "type": "string" },
59 "tag": { "type": "string" },
60 "limit": { "type": "integer", "minimum": 1, "maximum": MAX_LIMIT },
61 "offset": { "type": "integer", "minimum": 0 }
62 }
63 })
64 }
65 async fn call(&self, args: Value) -> Result<ToolCallResult> {
66 let limit = parse_limit(self.name(), args.get("limit"))?;
67 let offset = parse_offset(self.name(), args.get("offset"))?;
68
69 let tasks = self
70 .0
71 .tasks()
72 .list_all(self.0.user_id)
73 .await
74 .map_err(|e| fail(self.name(), e))?;
75
76 let project = args.get("project").and_then(Value::as_str);
77 let status = args
78 .get("status")
79 .and_then(Value::as_str)
80 .map(TaskStatus::from_str_or_default);
81 let tag = args.get("tag").and_then(Value::as_str);
82
83 let matched: Vec<&Task> = tasks
84 .iter()
85 .filter(|t| project.is_none_or(|p| t.project_name.as_deref() == Some(p)))
86 .filter(|t| status.as_ref().is_none_or(|s| &t.status == s))
87 .filter(|t| tag.is_none_or(|want| t.tags.iter().any(|have| have == want)))
88 .collect();
89
90 let total = matched.len();
91 let rows: Vec<Value> = matched
92 .into_iter()
93 .skip(offset)
94 .take(limit)
95 .map(task_summary_row)
96 .collect();
97
98 let mut reply = json!({
99 "count": rows.len(),
100 "total": total,
101 "offset": offset,
102 "tasks": rows,
103 });
104 // Only present when a page remains, so its absence is the stop condition.
105 let next = offset.saturating_add(reply["count"].as_u64().unwrap_or(0) as usize);
106 if next < total {
107 reply["next_offset"] = json!(next);
108 }
109
110 Ok(ToolCallResult::text(serde_json::to_string(&reply).unwrap()))
111 }
112 }
113
114 pub struct GetTask(pub Arc<Ctx>);
115
116 #[async_trait]
117 impl Tool for GetTask {
118 fn name(&self) -> &str {
119 "get_task"
120 }
121 fn description(&self) -> &str {
122 "Fetch one task by id, including its subtasks and annotations."
123 }
124 fn kind(&self) -> ToolKind {
125 ToolKind::Read
126 }
127 fn small_model_safe(&self) -> bool {
128 true
129 }
130 fn input_schema(&self) -> Value {
131 json!({
132 "type": "object",
133 "properties": { "id": { "type": "string" } },
134 "required": ["id"]
135 })
136 }
137 async fn call(&self, args: Value) -> Result<ToolCallResult> {
138 let id = parse_task_id(self.name(), req_str(self.name(), &args, "id")?)?;
139 let repo = self.0.tasks();
140 let task = repo
141 .get_by_id(id, self.0.user_id)
142 .await
143 .map_err(|e| fail(self.name(), e))?
144 .ok_or_else(|| Error::ToolFailed {
145 tool: self.name().to_string(),
146 message: format!("no task with id {id}"),
147 })?;
148
149 let subtasks = repo
150 .get_subtasks_for_task(id)
151 .await
152 .map_err(|e| fail(self.name(), e))?;
153 let annotations = repo
154 .get_annotations_for_task(id)
155 .await
156 .map_err(|e| fail(self.name(), e))?;
157
158 let mut row = task_row(&task);
159 row["subtasks"] = json!(
160 subtasks
161 .iter()
162 .map(|s| json!({ "text": s.text, "completed": s.is_completed }))
163 .collect::<Vec<_>>()
164 );
165 row["annotations"] = json!(
166 annotations
167 .iter()
168 .map(|a| json!({ "note": a.note, "created_at": a.timestamp.to_rfc3339() }))
169 .collect::<Vec<_>>()
170 );
171 Ok(ToolCallResult::text(serde_json::to_string(&row).unwrap()))
172 }
173 }
174
175 // writes
176
177 pub struct CreateTask(pub Arc<Ctx>);
178
179 #[async_trait]
180 impl Tool for CreateTask {
181 fn name(&self) -> &str {
182 "create_task"
183 }
184 fn description(&self) -> &str {
185 "Create one task. `project` (name) is resolved to a project, creating it if absent. `due` accepts RFC 3339 or YYYY-MM-DD; `priority` is High|Medium|Low."
186 }
187 fn kind(&self) -> ToolKind {
188 ToolKind::Write(caps::task_create())
189 }
190 fn input_schema(&self) -> Value {
191 json!({
192 "type": "object",
193 "properties": {
194 "description": { "type": "string" },
195 "project": { "type": "string" },
196 "tags": { "type": "array", "items": { "type": "string" } },
197 "due": { "type": "string" },
198 "priority": { "type": "string" }
199 },
200 "required": ["description"]
201 })
202 }
203 async fn call(&self, args: Value) -> Result<ToolCallResult> {
204 let description = req_str(self.name(), &args, "description")?.to_string();
205 let due = parse_due(self.name(), args.get("due"))?;
206 let priority = parse_priority(args.get("priority"));
207 let tags = parse_tags(args.get("tags"));
208
209 let project_id = match args.get("project").and_then(Value::as_str) {
210 Some(name) if !name.trim().is_empty() => Some(
211 ensure_project(&self.0, name)
212 .await
213 .map_err(|e| fail(self.name(), e))?,
214 ),
215 _ => None,
216 };
217
218 let mut builder = NewTask::builder(description).priority(priority).tags(tags);
219 if let Some(pid) = project_id {
220 builder = builder.project_id(pid);
221 }
222 if let Some(d) = due {
223 builder = builder.due(d);
224 }
225
226 let task = self
227 .0
228 .tasks()
229 .create(self.0.user_id, builder.build())
230 .await
231 .map_err(|e| fail(self.name(), e))?;
232
233 Ok(ToolCallResult::text(
234 serde_json::to_string(&json!({ "id": task.id.to_string() })).unwrap(),
235 ))
236 }
237 }
238
239 pub struct BulkImportTasks(pub Arc<Ctx>);
240
241 #[async_trait]
242 impl Tool for BulkImportTasks {
243 fn name(&self) -> &str {
244 "bulk_import_tasks"
245 }
246 fn description(&self) -> &str {
247 "Create many tasks in one call (the /dellm migration primitive). Each item: {description, project?, tags?, due?, priority?, source?}. `source` is a provenance key (e.g. file:line); an item whose source tag already exists is skipped, so re-running is idempotent. Returns created/skipped counts and the new task ids."
248 }
249 fn kind(&self) -> ToolKind {
250 ToolKind::Write(caps::task_bulk_import())
251 }
252 fn input_schema(&self) -> Value {
253 json!({
254 "type": "object",
255 "properties": {
256 "tasks": {
257 "type": "array",
258 "items": {
259 "type": "object",
260 "properties": {
261 "description": { "type": "string" },
262 "project": { "type": "string" },
263 "tags": { "type": "array", "items": { "type": "string" } },
264 "due": { "type": "string" },
265 "priority": { "type": "string" },
266 "source": { "type": "string" }
267 },
268 "required": ["description"]
269 }
270 }
271 },
272 "required": ["tasks"]
273 })
274 }
275 async fn call(&self, args: Value) -> Result<ToolCallResult> {
276 let items =
277 args.get("tasks")
278 .and_then(Value::as_array)
279 .ok_or_else(|| Error::InvalidArgs {
280 tool: self.name().to_string(),
281 message: "missing array field `tasks`".into(),
282 })?;
283
284 let repo = self.0.tasks();
285
286 // Existing source tags → idempotency. One scan up front.
287 let existing = repo
288 .list_all(self.0.user_id)
289 .await
290 .map_err(|e| fail(self.name(), e))?;
291 let mut seen_sources: HashSet<String> = existing
292 .iter()
293 .flat_map(|t| t.tags.iter())
294 .filter(|tag| tag.starts_with("source:"))
295 .cloned()
296 .collect();
297
298 let mut project_cache: HashMap<String, ProjectId> = HashMap::new();
299 let mut created_ids = Vec::new();
300 let mut skipped = 0usize;
301
302 for (idx, item) in items.iter().enumerate() {
303 let description = item
304 .get("description")
305 .and_then(Value::as_str)
306 .filter(|s| !s.trim().is_empty())
307 .ok_or_else(|| Error::InvalidArgs {
308 tool: self.name().to_string(),
309 message: format!("tasks[{idx}] is missing a non-empty `description`"),
310 })?
311 .to_string();
312
313 let mut tags = parse_tags(item.get("tags"));
314
315 // Provenance / idempotency.
316 if let Some(source) = item.get("source").and_then(Value::as_str) {
317 let tag = source_tag(source);
318 if !seen_sources.insert(tag.clone()) {
319 skipped += 1;
320 continue;
321 }
322 tags.push(tag);
323 }
324
325 let due = parse_due(self.name(), item.get("due"))?;
326 let priority = parse_priority(item.get("priority"));
327
328 let project_id = match item.get("project").and_then(Value::as_str) {
329 Some(name) if !name.trim().is_empty() => Some(
330 resolve_project_cached(&self.0, name, &mut project_cache)
331 .await
332 .map_err(|e| fail(self.name(), e))?,
333 ),
334 _ => None,
335 };
336
337 let mut builder = NewTask::builder(description).priority(priority).tags(tags);
338 if let Some(pid) = project_id {
339 builder = builder.project_id(pid);
340 }
341 if let Some(d) = due {
342 builder = builder.due(d);
343 }
344
345 let task = repo
346 .create(self.0.user_id, builder.build())
347 .await
348 .map_err(|e| fail(self.name(), e))?;
349 created_ids.push(task.id.to_string());
350 }
351
352 Ok(ToolCallResult::text(
353 serde_json::to_string(&json!({
354 "created": created_ids.len(),
355 "skipped": skipped,
356 "task_ids": created_ids,
357 }))
358 .unwrap(),
359 ))
360 }
361 }
362
363 pub struct UpdateTaskTool(pub Arc<Ctx>);
364
365 #[async_trait]
366 impl Tool for UpdateTaskTool {
367 fn name(&self) -> &str {
368 "update_task"
369 }
370 fn description(&self) -> &str {
371 "Update fields of an existing task. Only the fields you pass change; the rest keep their current values. Accepts `description`, `priority`, `due`, `status` (Pending|Started|Completed), `tags`, `project`, and `commit`. `commit` appends an advancing commit to the task's commit list, repo-qualified as `<repo>@<shortsha>` (e.g. `deox@7c236fca8`); use `complete_task` to record the closing commit."
372 }
373 fn kind(&self) -> ToolKind {
374 ToolKind::Write(caps::task_update())
375 }
376 fn input_schema(&self) -> Value {
377 json!({
378 "type": "object",
379 "properties": {
380 "id": { "type": "string" },
381 "description": { "type": "string" },
382 "priority": { "type": "string" },
383 "due": { "type": "string" },
384 "status": { "type": "string" },
385 "tags": { "type": "array", "items": { "type": "string" } },
386 "project": { "type": "string" },
387 "commit": { "type": "string" }
388 },
389 "required": ["id"]
390 })
391 }
392 async fn call(&self, args: Value) -> Result<ToolCallResult> {
393 let id = parse_task_id(self.name(), req_str(self.name(), &args, "id")?)?;
394 let repo = self.0.tasks();
395 let current = repo
396 .get_by_id(id, self.0.user_id)
397 .await
398 .map_err(|e| fail(self.name(), e))?
399 .ok_or_else(|| Error::ToolFailed {
400 tool: self.name().to_string(),
401 message: format!("no task with id {id}"),
402 })?;
403
404 // Start from the current task, overlay only the provided fields.
405 let mut patch = update_from_task(&current);
406 if let Some(desc) = args.get("description").and_then(Value::as_str) {
407 patch.description = desc.to_string();
408 }
409 if args.get("priority").is_some() {
410 patch.priority = parse_priority(args.get("priority"));
411 }
412 if let Some(due) = args.get("due") {
413 patch.due = parse_due(self.name(), Some(due))?;
414 }
415 if let Some(status) = args.get("status").and_then(Value::as_str) {
416 patch.status = TaskStatus::from_str_or_default(status);
417 }
418 if let Some(tags) = args.get("tags") {
419 patch.tags = parse_tags(Some(tags));
420 }
421 if let Some(name) = args.get("project").and_then(Value::as_str) {
422 patch.project_id = if name.trim().is_empty() {
423 None
424 } else {
425 Some(
426 ensure_project(&self.0, name)
427 .await
428 .map_err(|e| fail(self.name(), e))?,
429 )
430 };
431 }
432
433 let updated = repo
434 .update(id, self.0.user_id, patch)
435 .await
436 .map_err(|e| fail(self.name(), e))?
437 .ok_or_else(|| Error::ToolFailed {
438 tool: self.name().to_string(),
439 message: format!("task {id} vanished during update"),
440 })?;
441
442 // Record an advancing commit (if given) as a non-primary Pending token after
443 // the field update, then re-read so the returned row reflects it. Idempotent
444 // via the deterministic token id.
445 if let Some(commit) = args
446 .get("commit")
447 .and_then(Value::as_str)
448 .map(str::trim)
449 .filter(|s| !s.is_empty())
450 {
451 repo.record_status_token(
452 id,
453 self.0.user_id,
454 TOKEN_KIND_COMMIT,
455 commit,
456 TokenState::Pending,
457 false,
458 )
459 .await
460 .map_err(|e| fail(self.name(), e))?;
461 let refreshed = repo
462 .get_by_id(id, self.0.user_id)
463 .await
464 .map_err(|e| fail(self.name(), e))?
465 .unwrap_or(updated);
466 return Ok(ToolCallResult::text(
467 serde_json::to_string(&task_row(&refreshed)).unwrap(),
468 ));
469 }
470
471 Ok(ToolCallResult::text(
472 serde_json::to_string(&task_row(&updated)).unwrap(),
473 ))
474 }
475 }
476
477 pub struct CompleteTask(pub Arc<Ctx>);
478
479 #[async_trait]
480 impl Tool for CompleteTask {
481 fn name(&self) -> &str {
482 "complete_task"
483 }
484 fn description(&self) -> &str {
485 "Mark a task complete by id. Optional `commit` records the closing commit, repo-qualified as `<repo>@<shortsha>` (e.g. `deox@7c236fca8`); it is appended to the task's commit list and flagged as the closer."
486 }
487 fn kind(&self) -> ToolKind {
488 ToolKind::Write(caps::task_complete())
489 }
490 fn input_schema(&self) -> Value {
491 json!({
492 "type": "object",
493 "properties": {
494 "id": { "type": "string" },
495 "commit": { "type": "string" }
496 },
497 "required": ["id"]
498 })
499 }
500 async fn call(&self, args: Value) -> Result<ToolCallResult> {
501 let id = parse_task_id(self.name(), req_str(self.name(), &args, "id")?)?;
502 let repo = self.0.tasks();
503 repo.complete(id, self.0.user_id)
504 .await
505 .map_err(|e| fail(self.name(), e))?
506 .ok_or_else(|| Error::ToolFailed {
507 tool: self.name().to_string(),
508 message: format!("no task with id {id}"),
509 })?;
510
511 // Record the closing commit (if given) as the primary status token before
512 // re-reading, so the returned row carries it. A freshly-made commit is not
513 // yet pushed, so it starts Pending. Idempotent via the deterministic token id.
514 if let Some(commit) = args
515 .get("commit")
516 .and_then(Value::as_str)
517 .map(str::trim)
518 .filter(|s| !s.is_empty())
519 {
520 repo.record_status_token(
521 id,
522 self.0.user_id,
523 TOKEN_KIND_COMMIT,
524 commit,
525 TokenState::Pending,
526 true,
527 )
528 .await
529 .map_err(|e| fail(self.name(), e))?;
530 }
531
532 let completed = repo
533 .get_by_id(id, self.0.user_id)
534 .await
535 .map_err(|e| fail(self.name(), e))?
536 .ok_or_else(|| Error::ToolFailed {
537 tool: self.name().to_string(),
538 message: format!("task {id} vanished after completion"),
539 })?;
540 Ok(ToolCallResult::text(
541 serde_json::to_string(&task_row(&completed)).unwrap(),
542 ))
543 }
544 }
545
546 // helpers
547
548 /// Resolve a project by name, creating a default `SideProject` if absent.
549 async fn ensure_project(ctx: &Ctx, name: &str) -> goingson_core::repository::Result<ProjectId> {
550 let repo = ctx.projects();
551 if let Some(existing) = repo.find_by_name(ctx.user_id, name).await? {
552 return Ok(existing.id);
553 }
554 let created = repo
555 .create(
556 ctx.user_id,
557 NewProject {
558 name: name.to_string(),
559 description: String::new(),
560 project_type: ProjectType::SideProject,
561 status: ProjectStatus::default(),
562 },
563 )
564 .await?;
565 Ok(created.id)
566 }
567
568 /// `ensure_project` with a within-call cache, so a bulk import that references
569 /// the same project 200 times hits the database once.
570 async fn resolve_project_cached(
571 ctx: &Ctx,
572 name: &str,
573 cache: &mut HashMap<String, ProjectId>,
574 ) -> goingson_core::repository::Result<ProjectId> {
575 if let Some(id) = cache.get(name) {
576 return Ok(*id);
577 }
578 let id = ensure_project(ctx, name).await?;
579 cache.insert(name.to_string(), id);
580 Ok(id)
581 }
582
583 /// Build a full [`UpdateTask`] mirroring a task's current state, so a caller can
584 /// overlay just the fields it wants to change.
585 fn update_from_task(t: &Task) -> UpdateTask {
586 UpdateTask {
587 project_id: t.project_id,
588 milestone_id: t.milestone_id,
589 contact_id: t.contact_id,
590 description: t.description.clone(),
591 status: t.status.clone(),
592 priority: t.priority.clone(),
593 due: t.due,
594 tags: t.tags.clone(),
595 recurrence: t.recurrence.clone(),
596 recurrence_rule: t.recurrence_rule.clone(),
597 urgency: t.urgency,
598 scheduled_start: t.scheduled_start,
599 scheduled_duration: t.scheduled_duration,
600 estimated_minutes: t.estimated_minutes,
601 }
602 }
603