TealFlow

pip install tealtiger

import asyncio
from tealtiger.flow import TealFlow

async def main():
    # Load workflow from YAML
    flow = await TealFlow.from_file(".tealtiger/workflows/production.yaml")

    # Validate workflow
    validation = flow.validate()
    if not validation.valid:
        for error in validation.errors:
            print(f"[{error.type}] {error.job}: {error.message}")
        return

    # Execute workflow against a request
    result = await flow.execute(request, ctx)

    print(result.decision.action)       # Final merged decision
    print(result.execution_order)       # Actual execution order
    print(f"Duration: {result.duration_ms}ms")

asyncio.run(main())

# .tealtiger/workflows/production.yaml
name: production-governance
version: "1.0.0"
description: "Production governance workflow for customer-facing agents"

inherits: org://default-governance
environment: production

jobs:
  identity-check:
    module: nhi-governance
    config:
      enforceAttestation: true
      enforceScopes: true
    on_failure: deny

  secret-detection:
    module: teal-secrets
    config:
      enabled: true
      patterns: [aws, gcp, azure, github, generic]
    depends_on: [identity-check]
    on_failure: deny

  drift-detection:
    module: teal-drift
    config:
      enabled: true
      sensitivity: medium
    depends_on: [identity-check]
    on_failure: monitor

  evidence-emission:
    module: teal-proof
    config:
      receipts: true
      passport: true
    depends_on: [secret-detection, drift-detection]
    on_failure: monitor
    run: always

class TealFlow:
    @classmethod
    async def from_file(cls, path: str) -> "TealFlow": ...

    @classmethod
    def parse(cls, yaml_string: str) -> "TealFlow": ...

    def validate(self) -> ValidationResult: ...

    async def execute(self, request: dict, ctx: dict) -> FlowExecutionResult: ...

    def to_yaml(self) -> str: ...

jobs:
  A:
    module: teal-classifier
    # No dependencies — runs first

  B:
    module: teal-secrets
    depends_on: [A]  # Runs after A

  C:
    module: teal-drift
    depends_on: [A]  # Runs after A (parallel with B)

  D:
    module: teal-proof
    depends_on: [B, C]  # Runs after both B and C

Time →
  A ─────┐
         ├── B ────┐
         │         ├── D
         └── C ────┘

# org://default-governance (set by platform team)
name: org-default-governance
version: "2.0.0"
floor: true

jobs:
  identity-check:
    module: nhi-governance
    config:
      enforceAttestation: true
    on_failure: deny
    floor: true  # Cannot be weakened

  secret-detection:
    module: teal-secrets
    config:
      enabled: true
    on_failure: deny
    floor: true

flow = TealFlow.parse("""
name: team-workflow
inherits: org://default-governance

jobs:
  secret-detection:
    module: teal-secrets
    config:
      enabled: false  # Attempting to disable a floor job
""")

validation = flow.validate()
# validation.valid == False
# validation.errors[0].type == "FLOOR_VIOLATION"
# validation.errors[0].message == "Cannot weaken floor job: secret-detection"

from tealtiger.flow import WorkflowBuilder

workflow = (
    WorkflowBuilder("my-workflow")
    .version("1.0.0")
    .inherits("org://default-governance")
    .environment("production")
    .add_job("classifier", module="teal-classifier", config={"mode": "ensemble_union"}, on_failure="deny")
    .add_job("drift", module="teal-drift", config={"sensitivity": "high"}, depends_on=["classifier"], on_failure="monitor")
    .build()
)

# Export to YAML
yaml_str = workflow.to_yaml()

# Execute
result = await workflow.execute(request, ctx)

from dataclasses import dataclass

@dataclass
class FlowExecutionResult:
    decision: Decision                      # Final merged decision
    job_results: dict[str, JobResult]       # Per-job results
    execution_order: list[str]              # Actual execution order
    duration_ms: float                      # Total execution time
    workflow_version: str                   # Workflow version executed
    inheritance_chain: list[str]            # Inheritance chain

@dataclass
class JobResult:
    name: str
    status: str       # "success" | "failed" | "skipped" | "denied"
    decision: Decision | None
    error: str | None
    duration_ms: float

from tealtiger import TealEngine
from tealtiger.flow import TealFlow

engine = TealEngine(
    workflow=await TealFlow.from_file(".tealtiger/workflows/production.yaml"),
)

# evaluate() now executes the workflow
decision = await engine.evaluate(request, ctx)