Decision Trace Schema & API

[tfius]

Parent: #10 (EPIC: Transform DataCortex into Context Engine)
Priority: MEDIUM | Phase: 3 - Decision Layer | Complexity: Medium

What to Implement

Define and implement a standard schema for capturing "Decision Traces" - structured records of decisions including who made them, what inputs were considered, the outcome, and the reasoning.

Features

1. Decision Trace schema (YAML/JSON)
2. Decision storage in database
3. API for creating/querying decisions
4. CLI for logging decisions
5. Graph integration (Decision nodes linked to inputs/actors)

How to Implement

Step 1: Define Decision Schema

# src/datacortex/decisions/models.py
class DecisionStatus(str, Enum):
    proposed = "proposed"
    approved = "approved"
    rejected = "rejected"
    superseded = "superseded"

class DecisionInput(BaseModel):
    ref: str              # Reference URI (ref:issue:123)
    type: str             # issue, node, document, person, policy
    title: Optional[str] = None
    summary: Optional[str] = None

class Decision(BaseModel):
    id: str                              # dec-{uuid}
    timestamp: datetime
    actor: str                           # ref:person:sarah-chen
    title: str
    status: DecisionStatus = DecisionStatus.approved
    inputs: list[DecisionInput] = []
    outcome: str
    reasoning: str
    alternatives_considered: list[str] = []
    tags: list[str] = []
    supersedes: Optional[str] = None
    metadata: dict = {}

Example Decision

type: Decision
id: dec-123
timestamp: 2023-10-27T10:00:00Z
actor: ref:person:sarah-chen
title: Emergency deployment approval
inputs:
  - ref: ref:issue:123
    type: issue
    title: Critical bug in checkout
  - ref: ref:policy:v2
    type: policy
    title: Deployment Policy v2
outcome: Approved deployment of hotfix
reasoning: >
  Hotfix required for critical bug affecting 10% of users. 
  Bypassing standard 24h freeze due to revenue impact.
alternatives_considered:
  - Wait for standard deployment window
  - Rollback to previous version
tags:
  - deployment
  - hotfix
  - emergency

Step 2: Create Database Schema

CREATE TABLE decisions (
    id TEXT PRIMARY KEY,
    timestamp TEXT NOT NULL,
    actor TEXT NOT NULL,
    title TEXT NOT NULL,
    status TEXT NOT NULL,
    outcome TEXT NOT NULL,
    reasoning TEXT NOT NULL,
    alternatives_considered JSON,
    tags JSON,
    supersedes TEXT,
    metadata JSON,
    created_at TEXT,
    updated_at TEXT
);

CREATE TABLE decision_inputs (
    decision_id TEXT NOT NULL REFERENCES decisions(id),
    ref TEXT NOT NULL,
    type TEXT NOT NULL,
    title TEXT,
    summary TEXT,
    PRIMARY KEY (decision_id, ref)
);

CREATE INDEX idx_decisions_actor ON decisions(actor);
CREATE INDEX idx_decisions_timestamp ON decisions(timestamp);

Step 3: Create Decision Service

# src/datacortex/decisions/service.py
class DecisionService:
    def log_decision(self, decision: Decision) -> Decision:
        """Store a new decision."""

    def get_decision(self, decision_id: str) -> Optional[Decision]:
        """Get a decision by ID."""

    def search_decisions(
        self, actor: str = None, tags: list[str] = None, 
        since: datetime = None, limit: int = 50
    ) -> list[Decision]:
        """Search decisions with filters."""

    def get_decisions_for_input(self, ref: str) -> list[Decision]:
        """Get all decisions that reference a specific input."""

Step 4: Add API Endpoints

# src/datacortex/api/routes/decisions.py
@router.post("/")
async def log_decision(decision: Decision):

@router.get("/{decision_id}")
async def get_decision(decision_id: str):

@router.get("/")
async def search_decisions(actor: str = None, tag: str = None):

@router.get("/for/{ref}")
async def decisions_for_input(ref: str):

Step 5: Add CLI Commands

@cli.group()
def decision():
    """Manage decision traces."""
    pass

@decision.command('log')
@click.option('--actor', required=True)
@click.option('--title', required=True)
@click.option('--outcome', required=True)
@click.option('--reasoning', required=True)
@click.option('--input', 'inputs', multiple=True)
@click.option('--tag', 'tags', multiple=True)
def log_decision_cmd(actor, title, outcome, reasoning, inputs, tags):
    """Log a new decision."""

@decision.command('list')
def list_decisions():
    """List decisions."""

@decision.command('show')
@click.argument('decision_id')
def show_decision(decision_id):
    """Show decision details."""

Step 6: Graph Integration

def add_decisions_to_graph(graph: Graph, decisions: list[Decision]) -> Graph:
    """Add decision nodes and edges to the knowledge graph."""
    for d in decisions:
        graph.nodes.append(Node(
            id=d.id,
            title=d.title,
            type=NodeType.decision,
        ))
        # Link to actor
        graph.edges.append(Edge(source=d.actor, target=d.id, label='made_decision'))
        # Link to inputs
        for inp in d.inputs:
            graph.edges.append(Edge(source=d.id, target=inp.ref, label='considered'))

Acceptance Criteria

• Decision schema defined and validated
• Decisions stored in database
• datacortex decision log creates decisions via CLI
• API endpoints for CRUD operations
• Decisions appear as nodes in graph
• Decisions link to actors and inputs
• Documentation with examples

[joaquinariasco-lab]

Hi @tfius, this is a very thoughtful design for a Decision Trace layer.

The schema (actor, inputs, reasoning, outcome, supersedes, graph links) strongly resonates with a problem I’ve been working on in a separate project called Flowing. It focuses specifically on capturing structured decision traces in multi-agent LLM workflows, including parent/child relationships, tool invocations, reasoning context, and full execution spans for reproducibility.

Your Decision model abstraction is interesting because it generalizes beyond LLM agents into human/policy decisions. There’s a conceptual overlap:

• Structured decision nodes
• Explicit input references
• Actor attribution
• Graph integration
• Queryable trace storage

One thing I’ve been exploring is deterministic replay of agent workflows based on captured traces — essentially reconstructing why a system reached a certain outcome, step by step.

Curious whether you envision your Decision Layer eventually supporting:

• AI agent decision capture?
• Execution replay?
• Cross-decision lineage across workflows?

If there’s interest, I’d be happy to compare schema approaches, especially around trace semantics and parent-child decision modeling.

Project for context: https://github.com/joaquinariasco-lab/Flowing.git
Either way, solid direction on formalizing decision structure instead of relying on implicit logs.

Best,
Joaquin