Add auto-reverse functionality and enhance schedule models

robosystems/middleware/mcp/tools/schedule_tools.py

@@ -109,6 +109,10 @@ def get_tool_definition(self) -> dict[str, Any]:
             "type": "string",
             "description": "BS asset element for net book value tracking",
           },
+          "auto_reverse": {
+            "type": "boolean",
+            "description": "If true, closing entries auto-generate a reversing entry on the first day of the next period (default: false). Use for accruals that need to reverse.",
+          },
         },
         "required": [
           "name",
@@ -137,6 +141,7 @@ async def execute(self, arguments: dict[str, Any]) -> Any:
       credit_element_id=arguments["credit_element_id"],
       entry_type=arguments.get("entry_type", "closing"),
       memo_template=arguments.get("memo_template", "Monthly {structure_name}"),
+      auto_reverse=arguments.get("auto_reverse", False),
     )
 
     schedule_metadata = None
@@ -397,6 +402,8 @@ async def execute(self, arguments: dict[str, Any]) -> Any:
                 "amount": s.amount,
                 "status": s.status,
                 "entry_id": s.entry_id,
+                "reversal_entry_id": s.reversal_entry_id,
+                "reversal_status": s.reversal_status,
               }
               for s in status.schedules
             ],
@@ -494,7 +501,7 @@ async def execute(self, arguments: dict[str, Any]) -> Any:
         )
         session.commit()
 
-        return {
+        response = {
           "entry_id": result.entry_id,
           "status": result.status,
           "posting_date": str(result.posting_date),
@@ -512,6 +519,28 @@ async def execute(self, arguments: dict[str, Any]) -> Any:
             },
           ],
         }
+
+        if result.reversal:
+          response["reversal"] = {
+            "entry_id": result.reversal.entry_id,
+            "status": result.reversal.status,
+            "posting_date": str(result.reversal.posting_date),
+            "memo": result.reversal.memo,
+            "line_items": [
+              {
+                "element_id": result.reversal.debit_element_id,
+                "debit": result.reversal.amount,
+                "credit": 0,
+              },
+              {
+                "element_id": result.reversal.credit_element_id,
+                "debit": 0,
+                "credit": result.reversal.amount,
+              },
+            ],
+          }
+
+        return response
     except ValueError as exc:
       return {"error": str(exc)}
     except Exception as exc:

robosystems/models/api/extensions/schedules.py

@@ -1,5 +1,7 @@
 """Schedule request and response models."""
 
+from __future__ import annotations
+
 from datetime import date
 
 from pydantic import BaseModel, Field
@@ -18,6 +20,10 @@ class EntryTemplateRequest(BaseModel):
   memo_template: str = Field(
     "", description="Memo template ({structure_name} is replaced)"
   )
+  auto_reverse: bool = Field(
+    False,
+    description="Auto-generate a reversing entry on the first day of the next period",
+  )
 
 
 class ScheduleMetadataRequest(BaseModel):
@@ -86,6 +92,8 @@ class PeriodCloseItemResponse(BaseModel):
   amount: float
   status: str
   entry_id: str | None = None
+  reversal_entry_id: str | None = None
+  reversal_status: str | None = None
 
 
 class PeriodCloseStatusResponse(BaseModel):
@@ -105,6 +113,10 @@ class ClosingEntryResponse(BaseModel):
   debit_element_id: str
   credit_element_id: str
   amount: float
+  reversal: ClosingEntryResponse | None = None
+
+
+ClosingEntryResponse.model_rebuild()
 
 
 class ScheduleCreatedResponse(BaseModel):

robosystems/models/api/search.py

@@ -47,6 +47,7 @@ class SearchHit(BaseModel):
   document_id: str
   score: float
   source_type: str
+  parent_document_id: str | None = None
   entity_ticker: str | None = None
   entity_name: str | None = None
   section_label: str | None = None

robosystems/operations/documents/service.py

@@ -173,7 +173,7 @@ def _sync_to_opensearch(self, doc: Document) -> DocumentUploadResponse:
       logger.warning(f"Search service unavailable, skipping sync for {doc.id}")
       return DocumentUploadResponse(
         id=doc.id,
-        document_id=f"doc_{doc.graph_id}_{doc.id}",
+        document_id=f"udoc_{doc.id}",
         sections_indexed=0,
         total_content_length=len(doc.content),
         section_ids=[],
@@ -203,6 +203,6 @@ def _delete_from_opensearch(self, graph_id: str, document_id: str) -> None:
     if service is None:
       return
 
-    # The OpenSearch document_id prefix uses the PG document id as external_id
-    os_doc_id = f"doc_{graph_id}_{document_id}"
+    # The OpenSearch document_id prefix uses "udoc_" + PG document id
+    os_doc_id = f"udoc_{document_id}"
     service.delete_document(graph_id, os_doc_id)

robosystems/operations/schedules/service.py

@@ -37,6 +37,7 @@ class EntryTemplate:
   credit_element_id: str
   entry_type: str = "closing"
   memo_template: str = ""
+  auto_reverse: bool = False
 
 
 @dataclass
@@ -83,6 +84,8 @@ class PeriodCloseItem:
   amount: float  # dollars
   status: str  # "pending", "drafted", "posted"
   entry_id: str | None
+  reversal_entry_id: str | None = None
+  reversal_status: str | None = None
 
 
 @dataclass
@@ -108,6 +111,7 @@ class ClosingEntryResult:
   debit_element_id: str
   credit_element_id: str
   amount: float  # dollars
+  reversal: ClosingEntryResult | None = None
 
 
 # ── Service ──────────────────────────────────────────────────────────────
@@ -169,6 +173,7 @@ def create_schedule(
         "credit_element_id": entry_template.credit_element_id,
         "entry_type": entry_template.entry_type,
         "memo_template": entry_template.memo_template or f"Monthly schedule - {name}",
+        "auto_reverse": entry_template.auto_reverse,
       },
     }
     if schedule_metadata:
@@ -191,10 +196,13 @@ def create_schedule(
     session.flush()
 
     # Create associations to elements
+    # For schedules, from_element_id is the first element (debit) to anchor
+    # the presentation chain. to_element_id is the element being associated.
+    anchor_element_id = element_ids[0] if element_ids else None
     for i, element_id in enumerate(element_ids, 1):
       assoc = Association(
         structure_id=structure.id,
-        from_element_id=structure.id,
+        from_element_id=anchor_element_id,
         to_element_id=element_id,
         association_type="presentation",
         order_value=float(i),
@@ -369,22 +377,37 @@ def get_period_close_status(
           WHERE posting_date >= :period_start
             AND posting_date <= :period_end
             AND source_structure_id IS NOT NULL
+            AND type != 'reversing'
           ORDER BY source_structure_id,
             CASE status WHEN 'posted' THEN 1 WHEN 'draft' THEN 2 ELSE 3 END
+        ),
+        reversal AS (
+          SELECT DISTINCT ON (reversal_of)
+            reversal_of,
+            id AS reversal_entry_id,
+            status AS reversal_status
+          FROM entries
+          WHERE type = 'reversing'
+            AND reversal_of IS NOT NULL
+          ORDER BY reversal_of,
+            CASE status WHEN 'posted' THEN 1 WHEN 'draft' THEN 2 ELSE 3 END
         )
         SELECT
           s.id AS structure_id,
           s.name AS structure_name,
           s.metadata AS metadata,
           f.value AS amount,
           be.entry_id,
-          be.entry_status
+          be.entry_status,
+          r.reversal_entry_id,
+          r.reversal_status
         FROM structures s
         LEFT JOIN facts f ON f.structure_id = s.id
           AND f.period_start >= :period_start
           AND f.period_end <= :period_end
           AND f.element_id = (s.metadata->'entry_template'->>'debit_element_id')
         LEFT JOIN best_entry be ON be.source_structure_id = s.id
+        LEFT JOIN reversal r ON r.reversal_of = be.entry_id
         WHERE s.structure_type = 'schedule'
           AND s.is_active = true
         ORDER BY s.name
@@ -425,6 +448,8 @@ def get_period_close_status(
           amount=row.amount or 0.0,
           status=status,
           entry_id=row.entry_id,
+          reversal_entry_id=row.reversal_entry_id,
+          reversal_status=row.reversal_status,
         )
       )
 
@@ -557,6 +582,59 @@ def create_closing_entry(
 
     session.flush()
 
+    # Auto-reverse: create a reversing entry on the first day of the next period
+    reversal_result = None
+    if template.get("auto_reverse", False):
+      if period_end.month == 12:
+        reversal_date = date(period_end.year + 1, 1, 1)
+      else:
+        reversal_date = date(period_end.year, period_end.month + 1, 1)
+
+      reversal_memo = f"Reverse: {entry_memo}"
+
+      reversal_entry = Entry(
+        type="reversing",
+        status="draft",
+        posting_date=reversal_date,
+        memo=reversal_memo,
+        source_structure_id=structure_id,
+        reversal_of=entry.id,
+        created_by=created_by,
+      )
+      session.add(reversal_entry)
+      session.flush()
+
+      # Flipped DR/CR: debit element becomes credit, credit becomes debit
+      session.add(
+        LineItem(
+          entry_id=reversal_entry.id,
+          element_id=credit_element_id,
+          debit_amount=amount_cents,
+          credit_amount=0,
+          line_order=1,
+        )
+      )
+      session.add(
+        LineItem(
+          entry_id=reversal_entry.id,
+          element_id=debit_element_id,
+          debit_amount=0,
+          credit_amount=amount_cents,
+          line_order=2,
+        )
+      )
+      session.flush()
+
+      reversal_result = ClosingEntryResult(
+        entry_id=reversal_entry.id,
+        status="draft",
+        posting_date=reversal_date,
+        memo=reversal_memo,
+        debit_element_id=credit_element_id,
+        credit_element_id=debit_element_id,
+        amount=amount_dollars,
+      )
+
     return ClosingEntryResult(
       entry_id=entry.id,
       status="draft",
@@ -565,6 +643,7 @@ def create_closing_entry(
       debit_element_id=debit_element_id,
       credit_element_id=credit_element_id,
       amount=amount_dollars,
+      reversal=reversal_result,
     )
 
   # ── Private helpers ──────────────────────────────────────────────────

robosystems/operations/search/service.py

@@ -106,11 +106,18 @@ def search_documents(self, graph_id: str, request: SearchRequest) -> SearchRespo
         # so use section_label as minimal context
         snippet = source.get("section_label", "")
 
+      # Extract PG document ID for user docs: "udoc_{pg_doc_id}_{idx}" → pg_doc_id
+      doc_id = hit.get("_id", source.get("document_id", ""))
+      parent_doc_id = None
+      if doc_id.startswith("udoc_"):
+        parent_doc_id = doc_id.rsplit("_", 1)[0].removeprefix("udoc_") or None
+
       hits.append(
         SearchHit(
-          document_id=hit.get("_id", source.get("document_id", "")),
+          document_id=doc_id,
           score=hit.get("_score", 0.0),
           source_type=source.get("source_type", ""),
+          parent_document_id=parent_doc_id,
           entity_ticker=source.get("entity_ticker"),
           entity_name=source.get("entity_name"),
           section_label=source.get("section_label"),
@@ -193,10 +200,13 @@ def upload_document(
       raise ValueError("Document produced no indexable sections")
 
     # Generate base document ID
+    # User docs use "udoc_" prefix so the PG document ID is trivially extractable
+    # from search results: "udoc_{pg_doc_id}_{section_idx}" → rsplit("_", 1)[0][5:]
+    # SEC/pipeline docs use "doc_" prefix (no PG document table).
     if request.external_id:
-      base_id = f"doc_{graph_id}_{request.external_id}"
+      base_id = f"udoc_{request.external_id}"
     else:
-      base_id = f"doc_{graph_id}_{content_hash(request.content)}"
+      base_id = f"udoc_{content_hash(request.content)}"
 
     # Remove old sections if re-uploading
     self.client.delete_by_document_prefix(graph_id, base_id)

robosystems/routers/ledger/schedules.py

@@ -71,6 +71,7 @@ async def create_schedule(
         credit_element_id=body.entry_template.credit_element_id,
         entry_type=body.entry_template.entry_type,
         memo_template=body.entry_template.memo_template,
+        auto_reverse=body.entry_template.auto_reverse,
       )
 
       sm = None
@@ -247,6 +248,8 @@ async def get_period_close_status(
             amount=s.amount,
             status=s.status,
             entry_id=s.entry_id,
+            reversal_entry_id=s.reversal_entry_id,
+            reversal_status=s.reversal_status,
           )
           for s in status.schedules
         ],
@@ -297,6 +300,18 @@ async def create_closing_entry(
         None, mark_graph_stale, graph_id, "closing_entry_created"
       )
 
+      reversal_resp = None
+      if result.reversal:
+        reversal_resp = ClosingEntryResponse(
+          entry_id=result.reversal.entry_id,
+          status=result.reversal.status,
+          posting_date=result.reversal.posting_date,
+          memo=result.reversal.memo,
+          debit_element_id=result.reversal.debit_element_id,
+          credit_element_id=result.reversal.credit_element_id,
+          amount=result.reversal.amount,
+        )
+
       return ClosingEntryResponse(
         entry_id=result.entry_id,
         status=result.status,
@@ -305,6 +320,7 @@ async def create_closing_entry(
         debit_element_id=result.debit_element_id,
         credit_element_id=result.credit_element_id,
         amount=result.amount,
+        reversal=reversal_resp,
       )
 
   except ValueError as e: