Skip to content

Decision Logger Agent #15

@tfius

Description

@tfius

Parent: #10 (EPIC: Transform DataCortex into Context Engine)
Priority: MEDIUM | Phase: 3 - Decision Layer | Complexity: Large
Depends on: #13 (Identity Service), #14 (Decision Schema)

What to Implement

Create a background agent that monitors streams (file changes, commits, PR merges) for "decision-like" events and proposes logging them as formal Decision Traces.

Features

  1. Event monitoring for decision signals
  2. LLM-based decision detection and extraction
  3. Draft decision proposals for human review
  4. Approval workflow (accept/modify/reject)
  5. Automatic linking to related nodes

How to Implement

Step 1: Define Event Sources

# src/datacortex/decisions/agent/events.py
@dataclass
class Event:
    source: str          # git, file, slack
    type: str            # commit, file_change, message
    timestamp: datetime
    content: str
    metadata: dict

class EventSource(ABC):
    @abstractmethod
    async def poll(self) -> list[Event]:
        pass

class GitEventSource(EventSource):
    """Watch git commits for decision signals."""
    async def poll(self) -> list[Event]:
        # Parse commits, look for: "decided", "approved", "rejected"

class FileChangeEventSource(EventSource):
    """Watch for file changes that might indicate decisions."""
    # Watch patterns: "**/decisions/**", "**/ADR-*.md"

Step 2: Create Decision Detector

# src/datacortex/decisions/agent/detector.py
DETECTION_PROMPT = """
Analyze the following event and determine if it represents a decision.

Event: {event_content}

If this is a decision, extract:
1. Title
2. Actor (who made it)
3. Outcome
4. Reasoning
5. Inputs

Format as YAML with is_decision: true/false and confidence: 0.0-1.0
"""

class DecisionDetector:
    async def detect(self, event: Event) -> Optional[dict]:
        response = self.client.messages.create(
            model="claude-3-haiku-20240307",
            messages=[{"role": "user", "content": DETECTION_PROMPT.format(...)}]
        )
        return self._parse_response(response)

Step 3: Create Draft Store

# src/datacortex/decisions/agent/drafts.py
@dataclass
class DecisionDraft:
    id: str
    event: Event
    detected: dict
    status: str  # pending, approved, rejected, modified
    created_at: datetime
    final_decision_id: Optional[str] = None

class DraftStore:
    def save_draft(self, draft: DecisionDraft) -> None:
    def get_pending_drafts(self) -> list[DecisionDraft]:
    def approve_draft(self, draft_id: str, modifications: dict = None) -> Decision:
    def reject_draft(self, draft_id: str, reason: str = None) -> None:

Step 4: Create Agent Runner

# src/datacortex/decisions/agent/runner.py
class DecisionLoggerAgent:
    def __init__(self, config: dict):
        self.sources = self._init_sources(config)
        self.detector = DecisionDetector()
        self.draft_store = DraftStore(config['db_path'])
        self.poll_interval = config.get('poll_interval', 60)

    async def run(self):
        while True:
            for source in self.sources:
                events = await source.poll()
                for event in events:
                    detected = await self.detector.detect(event)
                    if detected and detected.get('confidence', 0) > 0.7:
                        draft = DecisionDraft(...)
                        self.draft_store.save_draft(draft)
                        await self._notify_draft(draft)
            await asyncio.sleep(self.poll_interval)

Step 5: Add CLI Commands

@decision.command('agent')
@click.option('--watch-git/--no-watch-git', default=True)
@click.option('--watch-files/--no-watch-files', default=False)
@click.option('--interval', default=60)
def run_agent(watch_git, watch_files, interval):
    """Run the decision logger agent."""

@decision.command('drafts')
def list_drafts():
    """List pending decision drafts."""

@decision.command('approve')
@click.argument('draft_id')
def approve_draft(draft_id):
    """Approve a decision draft."""

@decision.command('reject')
@click.argument('draft_id')
def reject_draft(draft_id):
    """Reject a decision draft."""

Acceptance Criteria

  • Agent runs in background monitoring events
  • Git commits analyzed for decision signals
  • File changes in decision directories detected
  • LLM extracts decision structure from events
  • Drafts stored for human review
  • CLI commands for reviewing/approving drafts
  • Approved drafts become formal decisions

Metadata

Metadata

Assignees

No one assigned

    Labels

    Type

    No type
    No fields configured for issues without a type.

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions