Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 36 additions & 23 deletions tapps_agents/workflow/enforcer.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@
from typing import Literal, TypedDict

from tapps_agents.core.llm_behavior import EnforcementConfig
from tapps_agents.workflow.intent_detector import IntentDetector, WorkflowType
from tapps_agents.workflow.message_formatter import MessageConfig, MessageFormatter

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -140,6 +142,14 @@ def __init__(
# Load config from file
self.config = self._load_config(config_path)

# Initialize intent detector and message formatter (ENH-001-S2, S3)
self._intent_detector = IntentDetector()
self._message_formatter = MessageFormatter(MessageConfig(
use_emoji=True,
show_benefits=self.config.suggest_workflows,
show_override=True,
))

logger.info(
f"WorkflowEnforcer initialized with mode={self.config.mode}, "
f"confidence_threshold={self.config.confidence_threshold}"
Expand Down Expand Up @@ -322,55 +332,58 @@ def _create_decision(
- allow: Empty message

Note:
Story 1 uses basic message templates.
Story 3 will add MessageFormatter for rich, context-aware messages.
Uses MessageFormatter (ENH-001-S3) for rich, context-aware messages.
Uses IntentDetector (ENH-001-S2) for workflow detection and confidence.
"""
# Determine should_block flag
should_block = action == "block" and self.config.block_direct_edits

# Generate message based on action
# Detect intent and get confidence (ENH-001-S2 integration)
detection_result = self._intent_detector.detect_workflow(
user_intent=user_intent,
file_path=file_path if file_path.exists() else None,
)
workflow = detection_result.workflow_type
confidence = detection_result.confidence

# Generate message using MessageFormatter (ENH-001-S3 integration)
if action == "block":
if self.config.suggest_workflows:
message = (
f"⚠️ Direct file edit blocked: {file_path}\n\n"
f"TappsCodingAgents workflows provide:\n"
f" • Automatic testing (80%+ coverage)\n"
f" • Quality gates (75+ score required)\n"
f" • Comprehensive documentation\n"
f" • Early bug detection\n\n"
f"Suggested workflow:\n"
f" @simple-mode *build \"{user_intent or 'Implement feature'}\"\n\n"
f"To bypass enforcement, use: --skip-enforcement flag"
message = self._message_formatter.format_blocking_message(
workflow=workflow,
user_intent=user_intent,
file_path=file_path,
confidence=confidence,
)
else:
message = (
f"⚠️ Direct file edit blocked: {file_path}\n"
f"Direct file edit blocked: {file_path}\n"
f"Use --skip-enforcement flag to bypass."
)
elif action == "warn":
if self.config.suggest_workflows:
message = (
f"💡 Consider using a workflow for: {file_path}\n"
f"Suggested: @simple-mode *build \"{user_intent or 'Implement feature'}\"\n"
f"(Proceeding with direct edit...)"
message = self._message_formatter.format_warning_message(
workflow=workflow,
user_intent=user_intent,
confidence=confidence,
)
else:
message = f"💡 Consider using a workflow for: {file_path}"
message = f"Consider using a workflow for: {file_path}"
else: # allow
message = ""
message = self._message_formatter.format_allow_message()

# Create decision
# Create decision with actual confidence from intent detection
decision: EnforcementDecision = {
"action": action,
"message": message,
"should_block": should_block,
"confidence": 0.0, # Story 1: No intent detection yet
"confidence": confidence,
}

# Log decision
logger.debug(
f"Enforcement decision: action={action}, should_block={should_block}, "
f"file_path={file_path}, user_intent={user_intent[:100]}"
f"file_path={file_path}, workflow={workflow.value}, confidence={confidence:.1f}%"
)

return decision
187 changes: 187 additions & 0 deletions tapps_agents/workflow/message_formatter.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,187 @@
"""
Message Formatter - ENH-001-S3

Formats enforcement messages for the Workflow Enforcement System.
Provides rich, context-aware messages for blocking and warning modes
with support for CLI and IDE output formats.

Design Principles:
- Separation of Concerns: Message formatting separate from enforcement logic
- Configurable: Emoji support, output format selection
- Context-Aware: Messages tailored to detected workflow type
"""

from __future__ import annotations

import logging
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Literal

from tapps_agents.workflow.intent_detector import WorkflowType

logger = logging.getLogger(__name__)


class OutputFormat(str, Enum):
"""Output format for messages."""

CLI = "cli"
IDE = "ide"


@dataclass
class MessageConfig:
"""Configuration for message formatting."""

use_emoji: bool = True
output_format: OutputFormat = OutputFormat.CLI
show_benefits: bool = True
show_override: bool = True


# Workflow-specific benefits
WORKFLOW_BENEFITS: dict[WorkflowType, list[str]] = {
WorkflowType.BUILD: [
"Automatic testing (80%+ coverage)",
"Quality gates (75+ score required)",
"Comprehensive documentation",
"Early bug detection",
],
WorkflowType.FIX: [
"Root cause analysis",
"Regression test generation",
"Fix verification",
"Related issue detection",
],
WorkflowType.REFACTOR: [
"Behavior preservation tests",
"Code quality improvement",
"Technical debt reduction",
"Safe incremental changes",
],
WorkflowType.REVIEW: [
"Security vulnerability detection",
"Code quality scoring",
"Best practice suggestions",
"Maintainability analysis",
],
}


class MessageFormatter:
"""
Formats enforcement messages for workflow enforcement system.

Provides rich, context-aware messages for blocking and warning modes
with configurable emoji support and output format selection.
"""

def __init__(self, config: MessageConfig | None = None) -> None:
"""Initialize formatter with optional configuration."""
self.config = config or MessageConfig()

def format_blocking_message(
self,
workflow: WorkflowType,
user_intent: str,
file_path: Path,
confidence: float,
) -> str:
"""
Format blocking mode message with benefits and override instructions.

Args:
workflow: Detected workflow type
user_intent: User's intent description
file_path: Path to file being edited
confidence: Detection confidence (0-100)

Returns:
Formatted blocking message
"""
emoji = self._get_emoji("block") if self.config.use_emoji else ""
intent_display = user_intent or "Implement feature"

lines = [
f"{emoji}Direct file edit blocked: {file_path}".strip(),
"",
f"Detected intent: {workflow.value} (confidence: {confidence:.0f}%)",
"",
]

if self.config.show_benefits:
lines.append("TappsCodingAgents workflows provide:")
for benefit in self._get_workflow_benefits(workflow):
bullet = " * " if self.config.output_format == OutputFormat.IDE else " - "
lines.append(f"{bullet}{benefit}")
lines.append("")

lines.append("Suggested workflow:")
lines.append(f' @simple-mode {workflow.value} "{intent_display}"')
lines.append("")

if self.config.show_override:
lines.extend(self._get_override_instructions())

return "\n".join(lines)

def format_warning_message(
self,
workflow: WorkflowType,
user_intent: str,
confidence: float,
) -> str:
"""
Format warning mode message (lighter suggestion).

Args:
workflow: Detected workflow type
user_intent: User's intent description
confidence: Detection confidence (0-100)

Returns:
Formatted warning message
"""
emoji = self._get_emoji("warn") if self.config.use_emoji else ""
intent_display = user_intent or "Implement feature"

lines = [
f"{emoji}Consider using a workflow (confidence: {confidence:.0f}%)".strip(),
f'Suggested: @simple-mode {workflow.value} "{intent_display}"',
"(Proceeding with direct edit...)",
]

return "\n".join(lines)

def format_allow_message(self) -> str:
"""Format allow message (empty for silent mode)."""
return ""

def _get_emoji(self, action: Literal["block", "warn", "allow"]) -> str:
"""Get emoji for action type."""
emojis = {
"block": "\u26a0\ufe0f ", # Warning sign
"warn": "\U0001f4a1 ", # Light bulb
"allow": "\u2705 ", # Check mark
}
return emojis.get(action, "")

def _get_workflow_benefits(self, workflow: WorkflowType) -> list[str]:
"""Get benefits list for workflow type."""
return WORKFLOW_BENEFITS.get(workflow, WORKFLOW_BENEFITS[WorkflowType.BUILD])

def _get_override_instructions(self) -> list[str]:
"""Get override instructions based on output format."""
if self.config.output_format == OutputFormat.IDE:
return [
"To bypass enforcement:",
" * Use --skip-enforcement flag",
" * Or set enforcement.mode: silent in config",
]
return [
"To bypass enforcement:",
" - Use --skip-enforcement flag",
" - Or set enforcement.mode: silent in .tapps-agents/config.yaml",
]
10 changes: 7 additions & 3 deletions tests/workflow/test_enforcer.py
Original file line number Diff line number Diff line change
Expand Up @@ -183,7 +183,7 @@ def test_blocking_mode_returns_block_action(
assert decision["action"] == "block"
assert decision["should_block"] is True
assert len(decision["message"]) > 0
assert decision["confidence"] == 0.0 # Story 1
assert decision["confidence"] >= 0.0 # IntentDetector provides confidence

def test_blocking_mode_message_includes_file_path(
self, blocking_config: EnforcementConfig
Expand Down Expand Up @@ -515,7 +515,9 @@ def test_decision_confidence_is_zero_for_story_1(
is_new_file=False,
)

assert decision["confidence"] == 0.0
# IntentDetector now provides actual confidence scores
assert decision["confidence"] >= 0.0
assert decision["confidence"] <= 100.0


# ============================================================================
Expand Down Expand Up @@ -690,7 +692,9 @@ def test_full_workflow_blocking_mode(self, temp_config_file: Path) -> None:
assert decision["action"] == "block"
assert decision["should_block"] is True
assert "workflow" in decision["message"].lower()
assert decision["confidence"] == 0.0
# IntentDetector now provides actual confidence scores
assert decision["confidence"] >= 0.0
assert decision["confidence"] <= 100.0

def test_multiple_sequential_calls(self, blocking_config: EnforcementConfig) -> None:
"""Test multiple sequential calls to enforcer."""
Expand Down
Loading
Loading