Integrando ao Sistema
Video da aula estara disponivel em breve
Da DSL ao Sistema Real
Na aula anterior, construimos o pipeline completo de compilacao: Lexer, Parser e Evaluator. Agora vamos integrar a DSL ao sistema harness.os, transformando regras textuais em governanca real. As tres etapas finais sao: compilacao para codigo Python, engine de regras e testes automatizados.
Diagrama
Texto DSL (.rules)
|
v
+------------------+
| DSL Compiler | Lexer + Parser + Evaluator
| (aula anterior) | Texto -> AST -> CompiledRules
+------------------+
|
v
+------------------+
| Code Generator | CompiledRules -> Python functions
| (esta aula) | Gera codigo executavel
+------------------+
|
v
+------------------+
| Rule Engine | Recebe eventos do MCP server
| (esta aula) | Avalia regras contra eventos
+------------------+ Retorna block/warn/require/log
|
v
+------------------+
| harness.os MCP | Integra com start_session, log_decision
| (producao) | Executa acoes das regras
+------------------+Code Generation: DSL para Python
Em vez de interpretar a AST a cada evento, podemos compilar as regras para funcoes Python. Isso e mais rapido e permite inspecao do codigo gerado.
Python
from typing import List
import textwrap
class DSLCompiler:
"""Compila regras da DSL para codigo Python."""
def __init__(self):
self.output_lines: List[str] = []
self.indent_level = 0
def emit(self, line: str):
indent = " " * self.indent_level
self.output_lines.append(f"{indent}{line}")
def compile(self, rules: List) -> str:
"""Gera um modulo Python com funcoes de regra."""
self.emit('"""Auto-generated rule functions from harness.os DSL."""')
self.emit('')
self.emit('from typing import Dict, Any, Optional')
self.emit('from dataclasses import dataclass')
self.emit('')
self.emit('@dataclass')
self.emit('class RuleResult:')
self.indent_level += 1
self.emit('rule_name: str')
self.emit('triggered: bool')
self.emit('action: str')
self.emit('message: str')
self.indent_level -= 1
self.emit('')
for rule in rules:
self._compile_rule(rule)
# Gerar funcao de registro
self._compile_registry(rules)
return '\n'.join(self.output_lines)
def _compile_rule(self, rule):
"""Gera funcao Python para uma regra."""
func_name = rule.name.replace('-', '_')
self.emit('')
self.emit(f'def check_{func_name}(event: Dict[str, Any]) -> Optional[RuleResult]:')
self.indent_level += 1
# Docstring
desc = ''
for p in rule.properties:
if p.key == 'description':
desc = p.value
self.emit(f'"""Regra: {rule.name} - {desc}"""')
# Trigger check
if rule.condition:
self.emit(f"if event.get('type') != '{rule.condition.trigger}':")
self.indent_level += 1
self.emit('return None')
self.indent_level -= 1
# Conditions
for cond in rule.condition.conditions:
self._compile_condition(cond)
# Action
action_type = rule.action.action_type if rule.action else 'warn'
message = ''
if rule.action:
for p in rule.action.properties:
if p.key == 'message':
message = p.value
self.emit(f"return RuleResult(")
self.indent_level += 1
self.emit(f"rule_name='{rule.name}',")
self.emit(f"triggered=True,")
self.emit(f"action='{action_type}',")
self.emit(f'message="{message}"')
self.indent_level -= 1
self.emit(")")
self.indent_level -= 1
def _compile_condition(self, cond):
"""Gera if-statement para uma condicao."""
if hasattr(cond, 'field') and hasattr(cond, 'value'):
if hasattr(cond, 'op') and cond.op: # ComparisonNode
self.emit(
f"if not (float(event.get('{cond.field}', 0)) "
f"{cond.op} {cond.value}):")
elif cond.field == 'contains':
self.emit(
f"if '{cond.value}' not in "
f"str(event.get('content', '')):")
elif cond.field == 'missing':
self.emit(
f"if '{cond.value}' in "
f"event.get('actions', []):")
else:
self.emit(
f"if event.get('{cond.field}') != '{cond.value}':")
self.indent_level += 1
self.emit('return None')
self.indent_level -= 1
def _compile_registry(self, rules):
"""Gera lista de todas as funcoes de regra."""
self.emit('')
self.emit('')
self.emit('# Registry: all rule check functions')
self.emit('RULE_CHECKS = [')
self.indent_level += 1
for rule in rules:
func_name = rule.name.replace('-', '_')
self.emit(f'check_{func_name},')
self.indent_level -= 1
self.emit(']')
# Exemplo: compilar regras para Python
# (usando ast_rules da aula anterior)
compiler = DSLCompiler()
python_code = compiler.compile(ast_rules)
print(python_code)
Saida (codigo Python gerado)
"""Auto-generated rule functions from harness.os DSL."""
from typing import Dict, Any, Optional
from dataclasses import dataclass
@dataclass
class RuleResult:
rule_name: str
triggered: bool
action: str
message: str
def check_commit_hygiene(event: Dict[str, Any]) -> Optional[RuleResult]:
"""Regra: commit-hygiene - Nunca adicionar Co-Authored-By"""
if event.get('type') != 'git_commit':
return None
if 'Co-Authored-By: Claude' not in str(event.get('content', '')):
return None
return RuleResult(
rule_name='commit-hygiene',
triggered=True,
action='block',
message="Remova o trailer Co-Authored-By do commit"
)
# Registry: all rule check functions
RULE_CHECKS = [
check_commit_hygiene,
]Rule Engine: Integracao com MCP
O HarnessRuleEngine conecta as regras compiladas ao MCP server do harness.os. Ele intercepta eventos, avalia regras e executa acoes.
Python
from dataclasses import dataclass, field
from typing import List, Dict, Any, Callable, Optional
from datetime import datetime
@dataclass
class RuleResult:
rule_name: str
triggered: bool
action: str
message: str
@dataclass
class AuditEntry:
timestamp: str
event_type: str
rule_name: str
action: str
message: str
class HarnessRuleEngine:
"""Engine de regras integrada ao harness.os MCP.
Responsabilidades:
1. Carregar regras compiladas (da DSL ou de Python)
2. Receber eventos do MCP (session_start, git_commit, etc.)
3. Avaliar regras contra eventos
4. Executar acoes (block, warn, require, log)
5. Manter audit trail
"""
def __init__(self):
self.rule_checks: List[Callable] = []
self.audit_log: List[AuditEntry] = []
self.blocked_count = 0
self.warned_count = 0
def load_rules(self, checks: List[Callable]):
"""Carrega funcoes de verificacao de regras."""
self.rule_checks = checks
print(f"[Engine] {len(checks)} regras carregadas")
def process_event(self, event: Dict[str, Any]) -> List[RuleResult]:
"""Processa um evento contra todas as regras."""
results = []
for check in self.rule_checks:
result = check(event)
if result and result.triggered:
results.append(result)
self._record_audit(event, result)
self._execute_action(result)
return results
def _execute_action(self, result: RuleResult):
"""Executa a acao definida pela regra."""
if result.action == 'block':
self.blocked_count += 1
print(f" [BLOCK] {result.rule_name}: {result.message}")
# Em producao: raise BlockedByRuleError(result)
elif result.action == 'warn':
self.warned_count += 1
print(f" [WARN] {result.rule_name}: {result.message}")
elif result.action == 'require':
print(f" [REQ] {result.rule_name}: {result.message}")
# Em producao: adiciona tarefa obrigatoria a sessao
elif result.action == 'log':
print(f" [LOG] {result.rule_name}: {result.message}")
# Em producao: insere em claude_session_events
def _record_audit(self, event: Dict, result: RuleResult):
"""Registra avaliacao no audit trail."""
self.audit_log.append(AuditEntry(
timestamp=datetime.now().isoformat(),
event_type=event.get('type', 'unknown'),
rule_name=result.rule_name,
action=result.action,
message=result.message
))
def get_stats(self) -> Dict:
return {
'rules_loaded': len(self.rule_checks),
'events_processed': len(self.audit_log),
'blocks': self.blocked_count,
'warnings': self.warned_count,
}
# --- Integrar tudo ---
# 1. Definir regras em DSL
dsl_rules = '''
rule "commit-hygiene" {
description "Nunca adicionar Co-Authored-By"
context all
priority high
when git_commit {
contains "Co-Authored-By: Claude"
}
then block {
message "Remova o trailer Co-Authored-By do commit"
}
}
rule "token-budget" {
description "Alertar quando contexto excede 80 porcento"
context all
priority medium
when context_load {
utilization > 0.8
}
then warn {
message "Contexto excedendo 80 porcento do budget"
}
}
'''
# 2. Compilar
lexer = DSLLexer(dsl_rules)
parser = DSLParser(lexer.tokenize())
ast = parser.parse()
evaluator = DSLEvaluator()
compiled = evaluator.compile_rules(ast)
# 3. Converter para check functions
def make_check(rule):
def check(event):
if rule.trigger and rule.trigger != event.get('type'):
return None
if rule.check and not rule.check(event):
return None
return RuleResult(rule.name, True, rule.action_type, rule.action_message)
return check
checks = [make_check(r) for r in compiled]
# 4. Carregar na engine
engine = HarnessRuleEngine()
engine.load_rules(checks)
# 5. Processar eventos
print("\n--- Evento 1: commit com Co-Authored-By ---")
engine.process_event({
'type': 'git_commit',
'content': 'Add feature\n\nCo-Authored-By: Claude Sonnet'
})
print("\n--- Evento 2: commit limpo ---")
engine.process_event({
'type': 'git_commit',
'content': 'Fix bug in parser'
})
print("\n--- Evento 3: contexto sobrecarregado ---")
engine.process_event({
'type': 'context_load',
'utilization': 0.92
})
print(f"\n--- Stats: {engine.get_stats()} ---")
Saida
[Engine] 2 regras carregadas
--- Evento 1: commit com Co-Authored-By ---
[BLOCK] commit-hygiene: Remova o trailer Co-Authored-By do commit
--- Evento 2: commit limpo ---
--- Evento 3: contexto sobrecarregado ---
[WARN] token-budget: Contexto excedendo 80 porcento do budget
--- Stats: {'rules_loaded': 2, 'events_processed': 2, 'blocks': 1, 'warnings': 1} ---Testes Automatizados
Todo compilador precisa de testes. Vamos criar uma suite de testes para cada componente da DSL:
Python
import unittest
class TestDSLLexer(unittest.TestCase):
def test_keywords(self):
lexer = DSLLexer('rule when then and or not')
tokens = lexer.tokenize()
types = [t.type for t in tokens[:-1]] # excluir EOF
self.assertEqual(types, [
TT.RULE, TT.WHEN, TT.THEN, TT.AND, TT.OR, TT.NOT
])
def test_string(self):
lexer = DSLLexer('"hello world"')
tokens = lexer.tokenize()
self.assertEqual(tokens[0].type, TT.STRING)
self.assertEqual(tokens[0].value, 'hello world')
def test_numbers(self):
lexer = DSLLexer('42 3.14')
tokens = lexer.tokenize()
self.assertEqual(tokens[0].value, '42')
self.assertEqual(tokens[1].value, '3.14')
def test_comparison_ops(self):
lexer = DSLLexer('> < >= <= == !=')
tokens = lexer.tokenize()
types = [t.type for t in tokens[:-1]]
self.assertEqual(types, [TT.GT, TT.LT, TT.GTE, TT.LTE, TT.EQ, TT.NEQ])
def test_comments_ignored(self):
lexer = DSLLexer('rule # this is a comment\n"name"')
tokens = lexer.tokenize()
self.assertEqual(tokens[0].type, TT.RULE)
self.assertEqual(tokens[1].type, TT.STRING)
def test_unterminated_string(self):
lexer = DSLLexer('"hello')
with self.assertRaises(SyntaxError):
lexer.tokenize()
class TestDSLParser(unittest.TestCase):
def _parse(self, source):
lexer = DSLLexer(source)
parser = DSLParser(lexer.tokenize())
return parser.parse()
def test_simple_rule(self):
rules = self._parse('''
rule "test" {
description "A test rule"
context all
}
''')
self.assertEqual(len(rules), 1)
self.assertEqual(rules[0].name, 'test')
self.assertEqual(len(rules[0].properties), 2)
def test_rule_with_when_then(self):
rules = self._parse('''
rule "commit-check" {
when git_commit {
contains "TODO"
}
then warn {
message "Commit contem TODO"
}
}
''')
self.assertIsNotNone(rules[0].condition)
self.assertEqual(rules[0].condition.trigger, 'git_commit')
self.assertIsNotNone(rules[0].action)
self.assertEqual(rules[0].action.action_type, 'warn')
def test_invalid_action_type(self):
with self.assertRaises(SyntaxError) as ctx:
self._parse('''
rule "bad" {
then explode {
message "boom"
}
}
''')
self.assertIn('explode', str(ctx.exception))
def test_multiple_rules(self):
rules = self._parse('''
rule "r1" { context all }
rule "r2" { context build }
rule "r3" { context testing }
''')
self.assertEqual(len(rules), 3)
class TestRuleEngine(unittest.TestCase):
def setUp(self):
self.engine = HarnessRuleEngine()
def check_coauthor(event):
if event.get('type') != 'git_commit':
return None
if 'Co-Authored-By' in event.get('content', ''):
return RuleResult('co-author', True,
'block', 'Remove trailer')
return None
self.engine.load_rules([check_coauthor])
def test_block_on_coauthor(self):
results = self.engine.process_event({
'type': 'git_commit',
'content': 'Fix\n\nCo-Authored-By: Claude'
})
self.assertEqual(len(results), 1)
self.assertEqual(results[0].action, 'block')
def test_pass_clean_commit(self):
results = self.engine.process_event({
'type': 'git_commit',
'content': 'Clean commit message'
})
self.assertEqual(len(results), 0)
def test_ignore_non_commit(self):
results = self.engine.process_event({
'type': 'session_start',
'content': 'Co-Authored-By: Claude'
})
self.assertEqual(len(results), 0)
def test_audit_trail(self):
self.engine.process_event({
'type': 'git_commit',
'content': 'Co-Authored-By: AI'
})
self.assertEqual(len(self.engine.audit_log), 1)
self.assertEqual(self.engine.audit_log[0].action, 'block')
# Para executar: python -m pytest test_dsl.py -vRecapitulacao do Curso
O que construimos: Ao longo de 6 modulos, percorremos todas as fases de um compilador -- e aplicamos cada conceito ao harness.os. Nesta ultima aula, fechamos o ciclo: uma DSL que usa lexer (Mod. 2), parser (Mod. 3), AST e semantica (Mod. 4), otimizacao (Mod. 5) e geracao de codigo (Mod. 5) para produzir governanca executavel.
Diagrama
Mapa do Curso -> Componentes da DSL:
Modulo 1: Introducao -> Entender o problema
Modulo 2: Analise Lexica -> DSLLexer (TT enum, Token, tokenize)
Modulo 3: Analise Sintatica -> DSLParser (recursive descent, AST nodes)
Modulo 4: Semantica -> DSLEvaluator (type checking, scoping)
Modulo 5: Geracao + Otimizacao -> DSLCompiler (Python code gen)
Modulo 6: Projeto -> HarnessRuleEngine (integracao)
Texto DSL Compiladores
+----+ +----+ +----+ clasicos:
| L | --> | P | --> | E |
+----+ +----+ +----+ Lexer -> Parser -> Semantic -> CodeGen
| | |
v v v
Tokens AST Rules Resultado: governanca executavel
para agentes AIResumo
- O DSLCompiler gera codigo Python a partir da AST, produzindo funcoes de verificacao independentes
- O HarnessRuleEngine conecta regras compiladas ao MCP, processando eventos e executando acoes
- Quatro tipos de acao (block, warn, require, log) cobrem diferentes niveis de governanca
- Audit trail registra toda avaliacao para rastreabilidade
- Testes automatizados cobrem lexer, parser e engine isoladamente
- O curso completo demonstra que tecnicas de compiladores sao aplicaveis a sistemas modernos de AI
Exercicio Final
Projeto completo: (1) Escreva pelo menos 5 regras na DSL cobrindo diferentes triggers e action types. (2) Compile-as para Python usando o DSLCompiler. (3) Integre no HarnessRuleEngine. (4) Escreva uma suite de testes com pelo menos 15 test cases cobrindo: lexer (tokens corretos, erros), parser (AST correto, erros de sintaxe), e engine (regras ativadas, nao ativadas, audit trail). (5) Bonus: adicione suporte a include para importar regras de outros arquivos e extends para heranca de regras.
Verifique seu entendimento
Qual e a vantagem de compilar regras DSL para funcoes Python em vez de interpretar a AST diretamente?