feat(desktop): add get_commitments, a read tool for what the person owes others

backend/database/action_items.py

@@ -1,4 +1,4 @@
 from dataclasses import dataclass, field
 from datetime import datetime, timezone, timedelta, tzinfo
 import logging
 from typing import Any, Dict, Iterable, List, Optional, Protocol, cast
@@ -715,6 +715,7 @@
     offset: int = 0,
     budget: Optional[ListReadBudget] = None,
     extra_fields: tuple[str, ...] = (),
+    scan_report: Optional[Dict[str, Any]] = None,
 ) -> List[Dict[str, Any]]:
     """
     Get action items for a user with optional filters.
@@ -734,6 +735,9 @@
     under its per-RPC timeout — so the aggregate read cannot outlive the
     request (#11831). Budget exhaustion stops at a bucket boundary and the
     route reports ``truncated``/``has_more`` honestly.
+    A caller that passes ``scan_report`` gets ``hit_limit`` (a filtered stream used
+    its whole document allowance, so soft-deleted rows may have hidden older live
+    ones) and ``docs_scanned`` written into it; the returned rows are unchanged.
     """
     offset = max(0, int(offset or 0))
     if limit is None or limit <= 0:
@@ -743,6 +747,7 @@
 
     need = min(offset + effective_limit, _ACTION_ITEMS_LIST_HARD_MAX)
     total_docs = 0
+    hit_limit = False
 
     def _base_query() -> Any:
         user_ref = db.collection('users').document(uid)
@@ -761,16 +766,16 @@
         return budget is not None and budget.truncated
 
     def _fetch_filtered(completed_filter: Optional[bool], row_budget: int) -> List[Dict[str, Any]]:
-        nonlocal total_docs
+        nonlocal total_docs, hit_limit
         if row_budget <= 0 or _out_of_budget():
             return []
         q = _base_query()
         if completed_filter is not None:
             q = q.where(filter=FieldFilter('completed', '==', completed_filter))
-        items, docs = _stream_action_items_bounded(
-            q, max_docs=_list_scan_budget(row_budget), budget=budget, extra_fields=extra_fields
-        )
+        max_docs = _list_scan_budget(row_budget)
+        items, docs = _stream_action_items_bounded(q, max_docs=max_docs, budget=budget, extra_fields=extra_fields)
         total_docs += docs
+        hit_limit = hit_limit or docs >= max_docs
         items.sort(key=_action_item_list_sort_key)
         return items[:row_budget]
 
@@ -806,6 +811,9 @@
         FirestoreReadMode.BOUNDED,
         total_docs,
     )
+    if scan_report is not None:
+        scan_report['hit_limit'] = hit_limit
+        scan_report['docs_scanned'] = total_docs
 
     if offset > 0:
         action_items = action_items[offset:]

backend/route_policy_manifest.yaml

@@ -13,6 +13,7 @@ routes:
   - {route_type: http, method: PATCH, path: "/v1/messaging/links/{link_id}", policy: {review_status: reviewed, auth: {mechanisms: [firebase_id_token], placement: dependency, scopes: []}, byok: validated_when_headers_present, rate_limit: {policy_name: none, key_subject: none, enforcement: none, placement: none}, timeout_class: default_method, surface: first_party_app, visibility: first_party, data_domain: chat, deprecation: {state: active}, owner: backend}}
   - {route_type: http, method: GET, path: /v1/dream/runs, policy: {review_status: reviewed, auth: {mechanisms: [firebase_id_token], placement: dependency, scopes: []}, byok: validated_when_headers_present, rate_limit: {policy_name: none, key_subject: none, enforcement: none, placement: none}, timeout_class: default_method, surface: first_party_app, visibility: first_party, data_domain: memories, deprecation: {state: active}, owner: backend}}
   - {route_type: http, method: POST, path: /v1/dream/runs, policy: {review_status: reviewed, auth: {mechanisms: [firebase_id_token], placement: dependency, scopes: []}, byok: validated_when_headers_present, rate_limit: {policy_name: none, key_subject: none, enforcement: none, placement: none}, timeout_class: sync_job, surface: first_party_app, visibility: first_party, data_domain: memories, deprecation: {state: active}, owner: backend}}
+  - {route_type: http, method: GET, path: /v1/tools/commitments, policy: {review_status: reviewed, auth: {mechanisms: [firebase_id_token], placement: dependency, scopes: []}, byok: validated_when_headers_present, rate_limit: {policy_name: "tools:search", key_subject: uid, enforcement: fail_open, placement: dependency}, timeout_class: default_method, surface: first_party_app, visibility: first_party, data_domain: action_items, deprecation: {state: active}, owner: backend}}
   - {route_type: http, method: POST, path: /v2/dream-agent/canary, policy: {review_status: reviewed, auth: {mechanisms: [service_oidc], placement: dependency, scopes: []}, byok: not_applicable, rate_limit: {policy_name: none, key_subject: none, enforcement: none, placement: none}, timeout_class: sync_job, surface: internal_task, visibility: internal, data_domain: memories, deprecation: {state: active}, owner: backend}}
   - route_type: http
     method: PUT

backend/routers/tools.py

@@ -13,6 +13,7 @@
 - GET   /v1/tools/action-items           — list action items
 - POST  /v1/tools/action-items           — create action item
 - PATCH /v1/tools/action-items/{id}      — update action item
+- GET   /v1/tools/commitments            — open commitments to other people (read-only)
 - POST  /v1/tools/calendar-events        — create calendar event
 """
 
@@ -36,6 +37,7 @@
     create_action_item_text,
     update_action_item_text,
 )
+from utils.retrieval.tool_services.commitments import get_commitments_result
 from utils.retrieval.tools.calendar_tools import create_calendar_event_tool
 
 logger = logging.getLogger(__name__)
@@ -74,6 +76,56 @@ class ToolResponse(BaseModel):
     sources: list[ToolSource] = Field(default_factory=list)
 
 
+class CommitmentAttribution(BaseModel):
+    names: list[str] = Field(default_factory=list)
+    person_ids: list[str] = Field(default_factory=list)
+    basis: Literal['speaker', 'speakers', 'conversation', 'none'] = 'none'
+    likely: bool = False
+
+
+class CommitmentItem(BaseModel):
+    task_id: str
+    description: str
+    due_at: Optional[str] = None
+    due_certainty: Optional[Literal['confirmed', 'tentative']] = None
+    created_at: Optional[str] = None
+    conversation_id: str
+    conversation_title: Optional[str] = None
+    conversation_started_at: Optional[str] = None
+    excerpt: Optional[str] = None
+    context: Optional[str] = None
+    attribution: CommitmentAttribution
+    partial_match: bool = False
+
+
+class CommitmentGroup(BaseModel):
+    names: list[str]
+    person_ids: list[str] = Field(default_factory=list)
+    basis: Literal['speaker', 'speakers', 'conversation']
+    likely: bool = False
+    partial_match: bool = False
+    open_count: int
+    commitments: list[CommitmentItem]
+
+
+class CommitmentTotals(BaseModel):
+    people: int = 0
+    commitments: int = 0
+    unattributed: int = 0
+    excluded_unknown_owner: int = 0
+    not_checked: int = 0
+
+
+class CommitmentsToolResponse(ToolResponse):
+    """ToolResponse plus the typed read the Mac maps directly; result_text is a plain fallback."""
+
+    since: Optional[str] = None
+    groups: list[CommitmentGroup] = Field(default_factory=list)
+    unattributed: list[CommitmentItem] = Field(default_factory=list)
+    totals: CommitmentTotals = Field(default_factory=CommitmentTotals)
+    truncated: bool = False
+
+
 def _ok(tool_name: str, text: str, sources: Optional[list[dict]] = None) -> dict:
     is_error = text.startswith("Error")
     return {
@@ -354,6 +406,28 @@ def update_action_item(
     return _ok("update_action_item", result)
 
 
+@router.get("/v1/tools/commitments", response_model=CommitmentsToolResponse)
+def get_commitments(
+    since: Optional[str] = Query(default=None, max_length=80, description="ISO date with timezone; max 90 days back"),
+    person: Optional[str] = Query(default=None, max_length=120, description="Only commitments to this person (name)"),
+    limit_people: int = Query(default=10, ge=1, le=20),
+    uid: str = Depends(with_rate_limit(get_current_user_uid, "tools:search")),
+):
+    """Read-only: never creates, edits or completes a task. Names only, never addresses."""
+    result = get_commitments_result(uid=uid, since=since, person=person, limit_people=limit_people)
+    envelope = _ok("get_commitments", result['result_text'], result['sources'])
+    if envelope['is_error']:
+        return envelope
+    return {
+        **envelope,
+        'since': result['since'],
+        'groups': result['groups'],
+        'unattributed': result['unattributed'],
+        'totals': result['totals'],
+        'truncated': result['truncated'],
+    }
+
+
 # --------------- calendar endpoints ---------------
 
 

backend/tests/support/firestore_query_driver_registry.py

@@ -1,4 +1,4 @@
 """Explicit driver registry for every statically discovered ``database.*`` querying function.
 
 Each entry is either a ``DriverEntry`` (driven across the full product of its
 shape-affecting parameter domains), a ``CoveredByEntry`` (query shapes are
@@ -67,6 +67,7 @@
 _PAGE = (25, 'scan/page bound; fixed small positive, not filter-affecting')
 _BUDGET = (None, 'request budget; RPC timeout/charge only, no filter effect')
 _EXTRA_FIELDS = (('provenance',), 'widens the select projection only; no filter or order effect')
+_SCAN_REPORT = (None, 'output-only dict the read fills with hit_limit/docs_scanned; no query effect')
 _DAYS = (30, 'window length; fixed, not filter-affecting')
 _NOW = (None, 'defaults to frozen clock; none/set produce identical filters')
 _NOOP = (noop, 'callback side-effect neutralized; does not affect filters')
@@ -278,7 +279,13 @@
             'due_start_date': [None, T0],
             'due_end_date': [None, T1],
         },
-        neutrals={'limit': _LIMIT_OPT, 'offset': _OFFSET, 'budget': _BUDGET, 'extra_fields': _EXTRA_FIELDS},
+        neutrals={
+            'limit': _LIMIT_OPT,
+            'offset': _OFFSET,
+            'budget': _BUDGET,
+            'extra_fields': _EXTRA_FIELDS,
+            'scan_report': _SCAN_REPORT,
+        },
     )
 )
 _add(

backend/tests/unit/test_action_items_pagination_order.py

@@ -182,3 +182,36 @@ def test_completed_normalized_to_bool(raw, expected_completed, expected_status):
     out = action_items.prepare_action_item_for_read(dict(raw))
     assert out['completed'] is expected_completed
     assert out['status'] == expected_status
+
+
+def _open_docs(count, deleted):
+    docs = []
+    for index in range(count):
+        doc = _Doc(f'd{index}', BASE - timedelta(minutes=index), None, completed=False)
+        if index < deleted:
+            doc._data['deleted'] = True
+        docs.append(doc)
+    return docs
+
+
+def test_scan_report_flags_a_stream_cut_by_soft_deleted_rows(fake_db):
+    # 40 open rows, the newest 35 soft-deleted: a page of 2 streams 2 + 32 slack
+    # documents, finds no live row, and older live rows stay unread.
+    fake_db.collection.return_value.document.return_value.collection.return_value = _Query(_open_docs(40, 35))
+    report = {}
+
+    items = action_items.get_action_items('uid1', completed=False, limit=2, scan_report=report)
+
+    assert items == []
+    assert report == {'hit_limit': True, 'docs_scanned': 34}
+
+
+def test_scan_report_is_clear_when_the_stream_ends_early_and_rows_are_unchanged(fake_db):
+    fake_db.collection.return_value.document.return_value.collection.return_value = _Query(_open_docs(5, 2))
+    report = {}
+
+    with_report = action_items.get_action_items('uid1', completed=False, limit=10, scan_report=report)
+    without_report = action_items.get_action_items('uid1', completed=False, limit=10)
+
+    assert report == {'hit_limit': False, 'docs_scanned': 5}
+    assert [item['id'] for item in with_report] == [item['id'] for item in without_report] == ['d2', 'd3', 'd4']