Shahzad Bhatti Welcome to my ramblings and rants!

timeline
    title AI Paradigms and Verification Approaches
    section Logic Era (1970s–1990s)
        1972 : Prolog — logic programming and knowledge representation
        1979 : Boyer-Moore theorem prover
        1986 : Eiffel introduces Design by Contract
        1987 : TLA created by Leslie Lamport
    section Hybrid Era (2000s–2010s)
        1999 : Z3 SMT solver (Microsoft Research)
        2005 : Alloy model finder
        2009 : Dafny created (Microsoft Research)
        2014 : TLA+ used at AWS for S3 and DynamoDB
    section LLM Era (2020s)
        2022 : ChatGPT and Copilot — probabilistic code generation goes mainstream
        2024 : Agentic coding — 10× code throughput becomes normal
        2025 : Spec-driven development movement emerges
        2026 : Formal verification as AI guardrail

has_role(alice, viewer).
has_role(bob, editor).
role_inherits(editor, viewer).
role_grants(viewer, read, docs).
role_grants(editor, write, docs).

can_access(User, Action, Resource) :-
    has_role(User, Role),
    role_grants(Role, Action, Resource).
can_access(User, Action, Resource) :-
    has_role(User, Role),
    role_inherits(Role, Parent),
    role_grants(Parent, Action, Resource).

?- can_access(bob, read, docs).
% Yes — bob is editor, editor inherits viewer, viewer grants read:docs.

Example: (A v B) ^ (¬A v C) ^ (¬B v ¬C)
SAT solver: A=true, B=false, C=true

(assert (= (+ x y) 10))
(assert (> x 3))
(assert (> y 3))
(check-sat)
--> sat; x=4, y=6

from ortools.sat.python import cp_model
model = cp_model.CpModel()
x = model.new_int_var(0, 10, 'x')
y = model.new_int_var(0, 10, 'y')
model.add(x + y == 10)
model.add(x > 3)
model.add(y > 3)
solver = cp_model.CpSolver()
solver.solve(model)  # finds x=4, y=6

Admin A: check(submitter) --> no conflict --> intend to assign
Admin B: check(approver)  --> no conflict --> intend to assign
Admin A: assign(submitter) OK
Admin B: assign(approver)   <-- SoD violated: both roles now held

(* Safety: SoD never violated *)
SoDOK ==
  ~(\E r1, r2 \in assigned : r1 # r2 /\ <<r1,r2>> \in Conflicts)

(* Liveness: every pending assignment eventually completes *)
Liveness ==
  \A a \in Admins :
    [](phase[a] = "checking" => <>(phase[a] \in {"done", "idle"}))

Model checking completed. No error has been found.
2349 states generated, 1046 distinct states found, 0 states left on queue.

java -jar ~/tla2tools.jar -config tla/RBACPolicyChange.cfg tla/RBACPolicyChange.tla

method Divide(a: int, b: int) returns (result: int)
  requires b != 0           // precondition: caller must ensure this
  ensures result * b == a   // postcondition: callee guarantees this
{
  return a / b;
}

predicate ValidStore(s: RBACStore) {
  // No dangling principal references
  && (forall id :: id in s.principals ==>
        s.principals[id].orgId in s.orgs)
  // Tenant isolation: role parents must be in same org
  && (forall rid :: rid in s.roles ==>
        (forall pid :: pid in SeqToSet(s.roles[rid].parentIds) ==>
           s.roles[pid].orgId == s.roles[rid].orgId))
  // Claim resources must exist in the store
  && (forall rid :: rid in s.roles ==>
        (forall c :: c in s.roles[rid].claims ==>
           c.resourceId in s.resources))
  // ... (8 more invariants)
}

lemma EmptyStoreIsValid()
  ensures ValidStore(EmptyStore())
{}

SOUNDNESS:    If Evaluate returns Allow, a valid claim chain EXISTS.
              No phantom permissions. No false positives.

COMPLETENESS: If a valid claim chain exists, Evaluate returns Allow.
              No missed permissions. No false negatives.

// The specification — states what "correct" means mathematically
predicate PolicySpec(req: Request, store: RBACStore, ctx: EvalContext) {
  var principal := store.principals[req.principalId];
  exists c :: c in PrincipalClaims(principal, store.roles) &&
              ClaimGrants(c, req.action, req.resourceId, ctx)
}

// The implementation — Dafny proves it matches PolicySpec for all inputs
method Evaluate(req: Request, store: RBACStore, ctx: EvalContext)
    returns (decision: Decision)
  requires ValidStore(store)
  requires req.principalId in store.principals
  requires store.principals[req.principalId].orgId == req.orgId
  ensures decision == Allow ==> PolicySpec(req, store, ctx)    // SOUNDNESS
  ensures decision == Deny  ==> !PolicySpec(req, store, ctx)   // COMPLETENESS

while i < |claimSeq|
  invariant decision == Deny ==>
    forall j :: 0 <= j < i ==>
      !ClaimGrants(claimSeq[j], req.action, req.resourceId, ctx)
  decreases |claimSeq| - i
{
  if ClaimGrants(claimSeq[i], req.action, req.resourceId, ctx) {
    decision := Allow;
    return;
  }
  i := i + 1;
}

lemma MoreClaimsMonotonic(req, store1, store2, ctx)
  requires store1 has subset of claims of store2
  ensures PolicySpec(req, store1, ctx) ==> PolicySpec(req, store2, ctx)

method AddParent(child: RoleId, parent: RoleId, roles: map<RoleId, Role>)
    returns (result: map<RoleId, Role>, ok: bool)
  requires NoCycles(roles)
  ensures ok  ==> NoCycles(result)    // DAG invariant always preserved
  ensures !ok ==> result == roles     // rejection leaves the store unchanged
{
  var wouldCycle := child in Ancestors(parent, roles, |roles|);
  if wouldCycle { return roles, false; }
  // safe to add the parent edge
}

predicate SoDSatisfied(assignedRoles: set<RoleId>, conflicts: ConflictSet) {
  forall a, b ::
    a in assignedRoles && b in assignedRoles && a != b ==>
      (a, b) !in conflicts
}

method AssignRole(principal, newRole, conflicts)
  requires SoDSatisfied(SeqToSet(principal.roleIds), conflicts)
  ensures ok  ==> SoDSatisfied(SeqToSet(updated.roleIds), conflicts)
  ensures !ok ==> exists existing ::
    existing in SeqToSet(principal.roleIds) &&
    (existing, newRole) in conflicts   // proof witness for why it was rejected

rbac_separation_of_duty.dfy(42,4): Error: a postcondition could not be proved
  ensures ok ==> SoDSatisfied(SeqToSet(updated.roleIds), conflicts)

Counterexample:
  principal.roleIds = ["approver"]
  newRole = "submitter"
  conflicts = {("submitter", "approver")}  ? (new, existing) direction missed

// Adding constraints can only reduce access, never increase it
lemma AddingConstraintReducesAccess(base, extra, ctx)
  ensures AllConstraintsHold(base + [extra], ctx) ==>
          AllConstraintsHold(base, ctx)

// Empty constraint list always passes (vacuous truth — no constraints = no restrictions)
lemma EmptyConstraintsAlwaysHold(ctx)
  ensures AllConstraintsHold([], ctx)
{}

// Higher usage makes quota constraints harder to satisfy
lemma HigherUsageHarder(limit, usage1, usage2, ctx)
  requires usage1 <= usage2
  ensures ConstraintHolds(MaxUsage(limit), ctx[usage:=usage2]) ==>
          ConstraintHolds(MaxUsage(limit), ctx[usage:=usage1])

// go/pkg/types/types.go

type Claim struct {
    ID          ClaimID
    Action      string
    ResourceID  ResID
    Constraints []Constraint   // empty = always passes (vacuous truth, proved by EmptyConstraintsAlwaysHold)
}

// NewTimeWindow enforces the Dafny precondition ValidConstraint at construction time
func NewTimeWindow(start, end int) (Constraint, error) {
    if start >= end || end > 24 {
        return Constraint{}, fmt.Errorf("invalid time window: start < end <= 24 required")
    }
    return Constraint{Kind: TimeWindowKind, StartHour: start, EndHour: end}, nil
}

// NewGeoFence — Dafny requires |regions| > 0
func NewGeoFence(regions []string) (Constraint, error) {
    if len(regions) == 0 {
        return Constraint{}, fmt.Errorf("geo fence requires at least one region")
    }
    return Constraint{Kind: GeoFenceKind, Regions: regions}, nil
}

// NewMaxUsage — Dafny requires limit > 0
func NewMaxUsage(limit int) (Constraint, error) {
    if limit <= 0 {
        return Constraint{}, fmt.Errorf("max usage limit must be positive")
    }
    return Constraint{Kind: MaxUsageKind, MaxCount: limit}, nil
}

// go/pkg/constraints/constraints.go

// Holds mirrors Dafny's ConstraintHolds predicate exactly.
// Every case corresponds to a branch in the Dafny match expression.
func Holds(c types.Constraint, ctx types.EvalContext) bool {
    switch c.Kind {
    case types.TimeWindowKind:
        // Dafny: ctx.currentHour >= c.startHour && ctx.currentHour < c.endHour
        return ctx.CurrentHour >= c.StartHour && ctx.CurrentHour < c.EndHour
    case types.GeoFenceKind:
        // Dafny: ctx.currentRegion in c.allowedRegions
        for _, r := range c.Regions {
            if r == ctx.CurrentRegion {
                return true
            }
        }
        return false
    case types.MaxUsageKind:
        // Dafny: ctx.currentUsage < c.maxCount
        return ctx.CurrentUsage < c.MaxCount
    default:
        return false
    }
}

// AllHold evaluates a conjunction of constraints.
// Dafny proved: AllConstraintsHold([], ctx) == true (EmptyConstraintsAlwaysHold)
// Dafny proved: AllConstraintsHold(base + [extra], ctx) ==> AllConstraintsHold(base, ctx)
func AllHold(cs []types.Constraint, ctx types.EvalContext) bool {
    for _, c := range cs {
        if !Holds(c, ctx) {
            return false
        }
    }
    return true
}

// go/pkg/hierarchy/hierarchy.go

// HasCycle returns true if adding parent to child would create a cycle.
// Mirrors Dafny: child in Ancestors(parent, roles, |roles|)
func (r *Resolver) HasCycle(child, parent types.RoleID) bool {
    visited := map[types.RoleID]bool{}
    queue := []types.RoleID{parent}
    for len(queue) > 0 {
        current := queue[0]
        queue = queue[1:]
        if current == child {
            return true
        }
        if visited[current] {
            continue
        }
        visited[current] = true
        if role, ok := r.roles[current]; ok {
            queue = append(queue, role.ParentIDs...)
        }
    }
    return false
}

// AddParent adds a parent role with cycle guard.
// Mirrors Dafny: requires NoCycles, ensures NoCycles preserved or store unchanged.
func (r *Resolver) AddParent(child, parent types.RoleID) error {
    if r.HasCycle(child, parent) {
        return fmt.Errorf("adding parent %s to %s would create a cycle", parent, child)
    }
    role := r.roles[child]
    role.ParentIDs = append(role.ParentIDs, parent)
    r.roles[child] = role
    return nil
}

// TransitiveClaims collects all claims reachable through the role hierarchy.
// BFS bounded by number of roles — same as Dafny's fuel parameter.
func TransitiveClaims(roleID types.RoleID, roles map[types.RoleID]types.Role) []types.Claim {
    var claims []types.Claim
    visited := map[types.RoleID]bool{}
    queue := []types.RoleID{roleID}
    for len(queue) > 0 {
        current := queue[0]
        queue = queue[1:]
        if visited[current] {
            continue
        }
        visited[current] = true
        role, ok := roles[current]
        if !ok {
            continue
        }
        claims = append(claims, role.Claims...)
        queue = append(queue, role.ParentIDs...)
    }
    return claims
}

// go/pkg/store/store.go

// AssignRole mirrors Dafny AssignRole:
//   requires SoDSatisfied(current roles, conflicts)
//   ensures  SoDSatisfied(updated roles, conflicts) OR rejection with witness
func (s *Store) AssignRole(principalID types.PrinID, roleID types.RoleID) error {
    s.mu.Lock()
    defer s.mu.Unlock()

    principal, ok := s.principals[principalID]
    if !ok {
        return fmt.Errorf("principal %s not found", principalID)
    }
    role, ok := s.roles[roleID]
    if !ok {
        return fmt.Errorf("role %s not found", roleID)
    }
    // Tenant isolation — from Dafny ValidStore predicate
    if role.OrgID != principal.OrgID {
        return fmt.Errorf("tenant isolation: role org %s != principal org %s",
            role.OrgID, principal.OrgID)
    }
    // SoD conflict check — checks BOTH directions, per Dafny SoDSatisfied predicate
    for _, existingRoleID := range principal.RoleIDs {
        if s.hasConflict(existingRoleID, roleID) {
            return fmt.Errorf("separation of duty: role %q conflicts with existing role %q",
                roleID, existingRoleID)
        }
    }
    // Safe to assign — SoD preserved (Dafny ensures clause holds)
    principal.RoleIDs = append(principal.RoleIDs, roleID)
    s.principals[principalID] = principal
    return nil
}

// go/pkg/policy/policy.go

// Evaluate decides Allow or Deny for a request.
// Preconditions from Dafny requires clauses: request fields valid, principal exists, tenant matches.
// Postconditions from Dafny ensures clauses: Allow iff valid claim chain exists.
func (e *Engine) Evaluate(req types.Request, ctx types.EvalContext) (types.Decision, error) {
    if err := req.Validate(); err != nil {
        return types.Deny, fmt.Errorf("invalid request: %w", err)
    }
    principal, ok := e.store.GetPrincipal(req.PrincipalID)
    if !ok {
        return types.Deny, nil  // deny-by-default — proved by DenyByDefault lemma
    }
    if principal.OrgID != req.OrgID {
        return types.Deny, fmt.Errorf("tenant isolation violated")
    }
    // Walk transitive claims — same BFS algorithm as Dafny Evaluate method
    claims := hierarchy.PrincipalClaims(principal, e.store.AllRoles())
    for _, c := range claims {
        if constraints.ClaimGrants(c, req.Action, req.ResourceID, ctx) {
            return types.Allow, nil
        }
    }
    return types.Deny, nil
}

// go/pkg/policy/policy_test.go

// Mirrors: DenyByDefault lemma in rbac_invariants.dfy
func TestProp_NoRolesAlwaysDenied(t *testing.T) {
    props := gopter.NewProperties(gopter.DefaultTestParameters())
    props.Property("principal with no roles is always denied", prop.ForAll(
        func(action, resource string) bool {
            s := store.New()
            _ = s.AddOrg(types.Organization{ID: "org", Name: "org"})
            _ = s.AddResource(types.Resource{ID: resource, Name: resource, Kind: "api"})
            _ = s.AddPrincipal(types.Principal{ID: "p", OrgID: "org", Name: "P"})
            engine := policy.New(s)
            req := types.Request{OrgID: "org", PrincipalID: "p",
                Action: action, ResourceID: resource}
            decision, _ := engine.Evaluate(req, types.EvalContext{CurrentHour: 12})
            return decision == types.Deny
        },
        gen.AlphaString(), gen.AlphaString(),
    ))
    props.TestingRun(t, gopter.NewFormatedReporter(false, 80, os.Stdout))
}

// Mirrors: AddingConstraintReducesAccess lemma in rbac_constraints.dfy
func TestProp_ConstraintMonotonicity(t *testing.T) {
    props := gopter.NewProperties(gopter.DefaultTestParameters())
    props.Property("subset of constraints passing implies prefix passes", prop.ForAll(
        func(hour int, region string, usage int) bool {
            ctx := types.EvalContext{
                CurrentHour:   abs(hour) % 24,
                CurrentRegion: region,
                CurrentUsage:  abs(usage) % 100,
            }
            tw, _ := types.NewTimeWindow(9, 17)
            base := []types.Constraint{tw}
            geo, _ := types.NewGeoFence([]string{"us-east-1"})
            extended := append(base, geo)
            // If the extended (stricter) set passes, the base set MUST also pass
            if constraints.AllHold(extended, ctx) {
                return constraints.AllHold(base, ctx)
            }
            return true
        },
        gen.Int(), gen.AnyString(), gen.Int(),
    ))
    props.TestingRun(t, gopter.NewFormatedReporter(false, 80, os.Stdout))
}

// Mirrors: HigherUsageHarder lemma in rbac_constraints.dfy
func TestProp_QuotaMonotonicity(t *testing.T) {
    props := gopter.NewProperties(gopter.DefaultTestParameters())
    props.Property("higher usage never turns Deny into Allow for quota", prop.ForAll(
        func(limit int, lo int, delta int) bool {
            limit = abs(limit)%100 + 1
            lo = abs(lo) % 200
            hi := lo + abs(delta)%100   // hi >= lo guaranteed

            quota, _ := types.NewMaxUsage(limit)
            ctxLo := types.EvalContext{CurrentUsage: lo}
            ctxHi := types.EvalContext{CurrentUsage: hi}

            // If quota passes at HIGHER usage, it MUST pass at lower usage
            if constraints.Holds(quota, ctxHi) {
                return constraints.Holds(quota, ctxLo)
            }
            return true
        },
        gen.Int(), gen.Int(), gen.Int(),
    ))
    props.TestingRun(t, gopter.NewFormatedReporter(false, 80, os.Stdout))
}

// Mirrors: OwnClaimsIncluded lemma in rbac_role_hierarchy.dfy
func TestProp_OwnClaimsAlwaysIncluded(t *testing.T) {
    props := gopter.NewProperties(gopter.DefaultTestParameters())
    props.Property("role's own claims always appear in transitive claims", prop.ForAll(
        func(claimCount int) bool {
            claimCount = abs(claimCount)%5 + 1
            var claims []types.Claim
            for i := 0; i < claimCount; i++ {
                claims = append(claims, types.Claim{
                    ID:         types.ClaimID(fmt.Sprintf("c%d", i)),
                    Action:     fmt.Sprintf("action%d", i),
                    ResourceID: "res1",
                })
            }
            roles := map[types.RoleID]types.Role{
                "role1": {ID: "role1", OrgID: "org1", Claims: claims},
            }
            transitive := hierarchy.TransitiveClaims("role1", roles)
            for _, c := range claims {
                found := false
                for _, tc := range transitive {
                    if tc.ID == c.ID {
                        found = true
                        break
                    }
                }
                if !found {
                    return false
                }
            }
            return true
        },
        gen.Int(),
    ))
    props.TestingRun(t, gopter.NewFormatedReporter(false, 80, os.Stdout))
}

+ empty constraint list always passes: OK, passed 200 tests.
+ hour in [start,end) passes, hour outside fails: OK, passed 500 tests.
+ subset of constraints passing implies prefix passes: OK, passed 300 tests.
+ higher usage never turns Deny into Allow for quota: OK, passed 300 tests.
+ role's own claims always appear in transitive claims: OK, passed 200 tests.
+ adding a parent never reduces transitive claims: OK, passed 200 tests.
+ unknown principal is always denied: OK, passed 200 tests.
+ principal with no roles is always denied: OK, passed 200 tests.
PASS — 2300+ property-based test cases executed.

datatype ConstraintKind =
    | TimeWindow(startHour: nat, endHour: nat)
    | GeoFence(allowedRegions: seq<string>)
    | MaxUsage(maxCount: nat)
    | RateLimit(requestsPerMinute: nat)  // NEW

predicate ValidConstraint(c: ConstraintKind) {
    match c
    case TimeWindow(s, e) => 0 <= s < e <= 24
    case GeoFence(regions) => |regions| > 0
    case MaxUsage(max) => max > 0
    case RateLimit(rpm) => rpm > 0   // must be positive
}

predicate ConstraintHolds(c: ConstraintKind, ctx: EvalContext) {
    match c
    case TimeWindow(s, e) => s <= ctx.currentHour < e
    case GeoFence(regions) => ctx.currentRegion in regions
    case MaxUsage(max) => ctx.currentUsage < max
    case RateLimit(rpm) => ctx.currentRequestRate < rpm   // NEW
}

// Lower rate limit is harder to satisfy — proved automatically
lemma LowerRateLimitHarder(rpm1: nat, rpm2: nat, ctx: EvalContext)
    requires rpm1 <= rpm2
    requires rpm1 > 0 && rpm2 > 0
    ensures ConstraintHolds(RateLimit(rpm1), ctx) ==>
            ConstraintHolds(RateLimit(rpm2), ctx)
{
    // Dafny proves this in under a second: if rate < rpm1 <= rpm2 then rate < rpm2
}

$ dafny verify dafny/rbac_constraints.dfy
Dafny program verifier finished with 12 verified, 0 errors

func NewRateLimit(rpm int) (Constraint, error) {
    if rpm <= 0 {
        return Constraint{}, fmt.Errorf("rate limit must be positive")
    }
    return Constraint{Kind: RateLimitKind, MaxCount: rpm}, nil
}

// Add to Holds() in constraints.go:
case types.RateLimitKind:
    return ctx.CurrentRequestRate < c.MaxCount

func TestProp_LowerRateLimitHarder(t *testing.T) {
    props := gopter.NewProperties(gopter.DefaultTestParameters())
    props.Property("lower rate limit is harder to satisfy", prop.ForAll(
        func(rpm1, rpm2, rate int) bool {
            rpm1 = abs(rpm1)%100 + 1
            rpm2 = rpm1 + abs(rpm2)%100   // rpm2 >= rpm1 guaranteed
            ctx := types.EvalContext{CurrentRequestRate: abs(rate) % 200}
            c1, _ := types.NewRateLimit(rpm1)
            c2, _ := types.NewRateLimit(rpm2)
            if constraints.Holds(c1, ctx) {
                return constraints.Holds(c2, ctx)
            }
            return true
        },
        gen.Int(), gen.Int(), gen.Int(),
    ))
    props.TestingRun(t, gopter.NewFormatedReporter(false, 80, os.Stdout))
}

# Step 1: Verify formal specs
$ make verify-dafny
[DAFNY] Verifying rbac_types.dfy...              ? PASS
[DAFNY] Verifying rbac_policy.dfy...             ? PASS
[DAFNY] Verifying rbac_role_hierarchy.dfy...     ? PASS
[DAFNY] Verifying rbac_separation_of_duty.dfy... ? PASS
[DAFNY] Verifying rbac_constraints.dfy...        ? PASS
[DAFNY] Verifying rbac_invariants.dfy...         ? PASS
All 6 Dafny files verified successfully.

# Step 2: Model check concurrent protocol
$ make check-tla
[TLA+] Model checking RBACPolicyChange...
Model checking completed. No error has been found.
  2349 states generated, 1046 distinct states found.

# Step 3: Run property-based tests
$ make test
+ empty constraint list always passes: OK, passed 200 tests.
+ time window boundary conditions: OK, passed 500 tests.
+ constraint monotonicity: OK, passed 300 tests.
+ quota monotonicity: OK, passed 300 tests.
+ own claims in transitive closure: OK, passed 200 tests.
+ adding parent never removes claims: OK, passed 200 tests.
+ unknown principal always denied: OK, passed 200 tests.
+ no roles always denied: OK, passed 200 tests.
PASS — 2300+ property-based test cases executed.

# Step 4: Smoke-test the API
$ make run &
Server starting on :9090...

# alice can read docs (viewer role, time window 9-17, currently hour 10)
$ curl -s localhost:9090/evaluate -d '{
  "org_id":"acme", "principal_id":"alice",
  "action":"read", "resource_id":"docs",
  "hour":10, "region":"us-east-1", "usage":0
}' | jq .decision
"Allow"

# alice denied at 10pm — time constraint blocks access outside 9-17
$ curl -s localhost:9090/evaluate -d '{
  "org_id":"acme", "principal_id":"alice",
  "action":"read", "resource_id":"reports",
  "hour":22, "region":"us-east-1", "usage":0
}' | jq .decision
"Deny"

# bob can write docs from US (editor role + geo constraint us-east-1)
$ curl -s localhost:9090/evaluate -d '{
  "org_id":"acme", "principal_id":"bob",
  "action":"write", "resource_id":"docs",
  "hour":12, "region":"us-east-1", "usage":0
}' | jq .decision
"Allow"

# bob denied from EU — geo constraint blocks non-US regions
$ curl -s localhost:9090/evaluate -d '{
  "org_id":"acme", "principal_id":"bob",
  "action":"write", "resource_id":"docs",
  "hour":12, "region":"eu-west-1", "usage":0
}' | jq .decision
"Deny"

# SoD blocks carol (finance role) from also being submitter
$ curl -s localhost:9090/principals/carol/roles -d '{"role_id":"submitter"}'
{"error":"separation of duty: role \"submitter\" conflicts with existing role \"finance\""}

// Dafny lemma — proved by the verifier
lemma EmptyConstraintsAlwaysHold(ctx: EvalContext)
  ensures AllConstraintsHold([], ctx)
{}

// Matching gopter property — catches Go translation bugs
props.Property("empty constraint list always passes", prop.ForAll(
    func(hour int, region string) bool {
        ctx := types.EvalContext{CurrentHour: hour % 24, CurrentRegion: region}
        return constraints.AllHold(nil, ctx)
    },
    gen.Int(), gen.AnyString(),
))

# Persist and restore session history across restarts
host.kv_put(f"session_history:{session_id}", json.dumps(messages))
raw = host.kv_get(f"session_history:{session_id}")
messages = json.loads(raw) if raw else []

# Index a skill under multiple trigger keywords
host.ts.write(["skill_trigger", "csv",         "skill-001"])
host.ts.write(["skill_trigger", "spreadsheet", "skill-001"])
host.ts.write(["skill_trigger", "pivot",       "skill-001"])

# Find every skill that might match — None is a wildcard
all_triggers = host.ts.read_all(["skill_trigger", None, None])
# ? [["skill_trigger","csv","skill-001"], ["skill_trigger","spreadsheet","skill-001"], ...]

# Audit log: all events of a specific type
events = host.ts.read_all(["audit", "tool_executed", None, None])

# Health snapshots: last N polls
snapshots = host.ts.read_all(["health_snapshot", None, None])

# Skill procedures can be several paragraphs — too large for KV values
host.blob.upload(f"skill_procedure_{skill_id}", procedure_text.encode())
procedure = host.blob.download(f"skill_procedure_{skill_id}").decode()

# Cron scheduler enqueues a job
host.channel.send("", "cron:pending", "cron_job", job_payload)

# Agent receives, processes, then acks — message redelivered if agent crashes before ack
msg, ok, _ = host.channel.receive("", "cron:pending", timeout_ms=5000)
if ok:
    # ... process the job ...
    host.channel.ack("", "cron:pending", msg["msg_id"])
    # or: host.channel.nack("", "cron:pending", msg["msg_id"], True)  # requeue

// Go — CronSchedulerActor.tick()
// TryAcquire returns false immediately if another node holds the lock
// TTL of 90s is longer than the 60s tick interval, preventing gaps
acquired, _ := host.Lock().TryAcquire("minihermes", "cron_leader", 90000)
if !acquired {
    return // another node is the leader this cycle
}
// Safe to fire jobs — only this node runs this block right now

@init_handler
def on_init(self, config: dict) -> None:
    host.process_groups.join("svc:health_monitor")
    # Arm the first tick — no external cron daemon needed
    host.send_after(self.poll_interval_ms, "poll_tick", {"op": "poll_tick"})

@handler("poll_tick", "cast")
def poll_tick(self) -> None:
    # ... do poll work ...
    # Re-arm: each tick schedules the next
    host.send_after(self.poll_interval_ms, "poll_tick", {"op": "poll_tick"})

# host.ask() — blocks until a response arrives (or timeout)
llm_resp = host.ask(llm_id, "completion",
                    {"messages": messages, "tools": tools},
                    timeout_ms=30000)

# host.send() — returns immediately, caller never waits
host.send(audit_id, "log_event",
          {"event_type": "tool_executed", "detail": f"tool={name}"})

# Increment a named counter — visible to monitoring without any external metrics system
host.incr_counter("llm_completions_total", 1)
host.incr_counter("tool_executions_total", 1)
host.incr_counter(f"tool_{name}_total", 1)
host.incr_counter("skill_matches_total", len(matched_ids))

# app-config.toml
[supervisor]
strategy = "one_for_one"           # restart ONLY the crashed child
max_restarts = 10
max_restart_window_seconds = 60    # if 10 crashes in 60s, escalate to parent supervisor

// wit/plexspaces-actor/host.wit
// Actors can ONLY call these imports — nothing else is accessible
interface host {
    send:       func(to: string, msg-type: string, payload: payload) -> result<_, actor-error>;
    ask:        func(to: string, msg-type: string, payload: payload, timeout-ms: u64) -> result<payload, actor-error>;
    kv-get:     func(key: string) -> result<payload, actor-error>;
    kv-put:     func(key: string, value: payload) -> result<_, actor-error>;
    http-fetch: func(link-name: string, method: string, path: string, request: payload) -> result<payload, actor-error>;
    ts-write:   func(tuple: list<string>) -> result<_, actor-error>;
    ts-read-all:func(pattern: list<option<string>>) -> result<list<list<string>>, actor-error>;
    // No filesystem. No env vars. No raw network. No process exec.
}

# Framework-enforced key scoping — no application code can bypass this
KV:          tenant-acme:prod:session_history:sess-001
TupleSpace:  tenant-acme:prod:["skill_trigger", "csv", "skill-001"]
PG:          tenant-acme:prod:svc:agent

User: "calculate 42 * 17 and remember the result"

  1. Restore session history from KV (survives restarts)
  2. Ask ContextCompressorActor: token budget > 75%?
     --> Yes: summarize the middle, keep the recent tail, archive original
  3. Ask SkillStoreActor: known procedures for "calculate" + "memory_store"?
     --> Found: inject skill into system prompt
  4. Ask ToolExecutorActor: list current tool schemas
  5. LOOP (max 8 iterations):
     a. Ask LLMGatewayActor: complete with these messages + tools
     b. stop_reason = tool_use:
        --> GuardrailsGate.check("calculator")   --> allow
        --> ToolExecutor.execute("calculator", {expr: "42*17"}) --> {result: 714}
        --> GuardrailsGate.check("memory_store") --> allow
        --> ToolExecutor.execute("memory_store", {key: "last_calc", value: "714"})
        --> Append results; continue loop
     c. stop_reason = end_turn --> break
  6. KV.put("session_history:sess-001", messages)   --  durable checkpoint
  7. send (fire-and-forget): SkillStoreActor.evaluate_for_learning
  8. send (fire-and-forget): AuditEventActor.log_event
  == "42 × 17 = 714. I've stored the result in your memory."

@actor
class AgentActor:
    system_prompt: str = state(default="You are a helpful AI assistant with access to tools.")
    messages: list     = state(default_factory=list)
    max_iterations: int = state(default=8)
    token_budget: int   = state(default=4096)

    @init_handler
    def on_init(self, config: dict) -> None:
        args = config.get("args", {})
        self.system_prompt = args.get("system_prompt", self.system_prompt)
        host.process_groups.join("svc:agent")
        # Publish capabilities for registry-based discovery
        host.registry.register(ctx="", object_type="actor", object_id=config["actor_id"],
                                object_category="agent",
                                capabilities=["chat", "tool_use", "memory"])

    @handler("chat")
    def chat(self, message: str = "", session_id: str = "") -> dict:
        # 1. Restore durable session
        if session_id:
            raw = host.kv_get(f"session_history:{session_id}")
            if raw:
                self.messages = json.loads(raw)
        self.messages.append({"role": "user", "content": message})

        # 2. Compress if over token budget
        comp_id, _ = pg_first("svc:context_compressor")
        if comp_id:
            resp = ask(comp_id, "check_and_compress",
                       {"messages": self.messages, "token_budget": self.token_budget})
            if resp and resp.get("compressed"):
                self.messages = resp["messages"]

        # 3. Inject matching skills
        skill_id, _ = pg_first("svc:skill_store")
        skill_context = ""
        if skill_id:
            resp = ask(skill_id, "match_skills", {"query": message})
            if resp and resp.get("skills"):
                skill_context = self._format_skills(resp["skills"])

        # 4. Get live tool schemas
        tool_exec_id, _ = pg_first("svc:tool_executor")
        tools = []
        if tool_exec_id:
            resp = ask(tool_exec_id, "list_tools", {})
            tools = resp.get("tools", []) if resp else []

        system = self.system_prompt
        if skill_context:
            system += f"\n\n## Relevant Skills\n{skill_context}"

        # 5. The tool loop — max_iterations prevents runaway execution
        final_response = ""
        for iteration in range(self.max_iterations):
            llm_id, _ = pg_first("svc:llm_gateway")
            llm_resp = ask(llm_id, "completion",
                           {"messages": [{"role": "system", "content": system}] + self.messages,
                            "tools": tools},
                           timeout_ms=30000)

            response    = llm_resp.get("response", {})
            stop_reason = response.get("stop_reason", "end_turn")
            self.messages.append({"role": "assistant",
                                   "content": response.get("content", ""),
                                   "stop_reason": stop_reason})

            if stop_reason == "end_turn":
                final_response = response.get("content", "")
                break

            if stop_reason == "tool_use":
                guard_id, _ = pg_first("svc:guardrails")
                for tc in response.get("tool_calls", []):
                    # Every tool call clears the guardrail first
                    if guard_id:
                        check = ask(guard_id, "check_tool",
                                    {"tool_name": tc["name"], "input": tc["input"]})
                        if check and check.get("decision") == "deny":
                            self.messages.append({"role": "tool",
                                                   "content": f"[denied: {tc['name']}]"})
                            continue
                    result = ask(tool_exec_id, "execute",
                                 {"name": tc["name"], "input": tc["input"]})
                    self.messages.append({"role": "tool",
                                          "tool_call_id": tc["id"],
                                          "content": json.dumps(result)})
                    host.send(audit_id, "log_event",
                              {"event_type": "tool_executed",
                               "detail": f"tool={tc['name']} session={session_id}"})
                    host.incr_counter("tool_executions_total", 1)

        # 6. Checkpoint session — durable across restarts
        if session_id:
            host.kv_put(f"session_history:{session_id}", json.dumps(self.messages))

        # 7+8. Async learning and audit — never block the response
        if skill_id:
            host.send(skill_id, "evaluate_for_learning",
                      {"messages": self.messages, "user_intent": message})
        host.incr_counter("agent_chats_total", 1)
        return {"status": "ok", "response": final_response, "session_id": session_id}

# Switch from Ollama to Anthropic — takes effect immediately, no restart
curl -X POST http://localhost:8091/api/v1/actors/llm_gateway/switch_provider \
  -d '{"provider":"anthropic","model":"claude-opus-4-8"}'

# Or to OpenAI
curl -X POST http://localhost:8091/api/v1/actors/llm_gateway/switch_provider \
  -d '{"provider":"openai","model":"gpt-4o"}'

@actor
class LLMGatewayActor:
    provider:              str  = state(default="ollama")
    model:                 str  = state(default="llama3.2")
    circuit_open:          bool = state(default=False)
    consecutive_failures:  int  = state(default=0)
    total_completions:     int  = state(default=0)

    @init_handler
    def on_init(self, config: dict) -> None:
        host.process_groups.join("svc:llm_gateway")
        host.send_after(30_000, "timer_tick", {"op": "timer_tick"})

    @handler("completion")
    def completion(self, messages: list = None, tools: list = None) -> dict:
        if self.circuit_open:
            # Fail fast — don't queue work behind a broken provider
            return {"status": "ok", "response": self._simulated_response(),
                    "circuit_open": True}
        try:
            result = self._call_provider(messages or [], tools or [])
            self.consecutive_failures = 0
            self.total_completions += 1
            host.incr_counter("llm_completions_total", 1)
            return {"status": "ok", "response": result}
        except Exception as e:
            self.consecutive_failures += 1
            if self.consecutive_failures >= 3:
                self.circuit_open = True
                host.warn(f"LLM circuit opened after {self.consecutive_failures} failures")
                host.incr_counter("llm_circuit_opens_total", 1)
            return {"error": str(e), "response": self._simulated_response()}

    @handler("timer_tick", "cast")
    def timer_tick(self) -> None:
        # Gradual recovery: one fault cleared per 30s tick
        # 3 faults ? 90s before circuit closes again — prevents flapping
        if self.circuit_open and self.consecutive_failures > 0:
            self.consecutive_failures -= 1
            if self.consecutive_failures == 0:
                self.circuit_open = False
                host.info("LLM circuit closed — provider available again")
        host.send_after(30_000, "timer_tick", {"op": "timer_tick"})

    @handler("switch_provider")
    def switch_provider(self, provider: str = "", model: str = "") -> dict:
        self.provider = provider
        self.model    = model
        # Switching resets the circuit — assume the new provider is healthy
        self.circuit_open         = False
        self.consecutive_failures = 0
        return {"status": "ok", "provider": provider, "model": model}

    def _call_provider(self, messages: list, tools: list) -> dict:
        if self.provider == "ollama":
            resp = host.http_fetch("ollama", "POST", "/api/chat",
                                   {"model": self.model, "messages": messages, "stream": False})
        elif self.provider == "anthropic":
            resp = host.http_fetch("anthropic", "POST", "/v1/messages",
                                   {"model": self.model, "messages": messages,
                                    "tools": tools, "max_tokens": 4096})
        elif self.provider == "openai":
            resp = host.http_fetch("openai", "POST", "/v1/chat/completions",
                                   {"model": self.model, "messages": messages, "tools": tools})
        return self._normalize(resp)

{
  "content":    "42 × 17 = 714",
  "stop_reason": "end_turn",
  "tool_calls": [],
  "usage":      {"input_tokens": 112, "output_tokens": 18}
}

@workflow_actor
class SkillExtractionWorkflow:

    @run_handler
    def run(self, payload: dict = None) -> dict:
        user_intent  = payload.get("user_intent", "")
        tool_sequence = payload.get("tool_sequence", [])
        domain        = payload.get("domain", "general")
        llm_id        = payload.get("llm_id", "")

        # Three focused LLM passes — each optimizes for a different extraction goal.
        # Python runs them sequentially (shared LLM budget).
        # Go runs them in true parallel goroutines for lower latency.
        name_result      = self._analyse_name(llm_id, user_intent, tool_sequence)
        # ? workflow checkpoints here; crash-safe from this point

        procedure_result = self._analyse_procedure(llm_id, user_intent, tool_sequence)
        # ? checkpoint

        trigger_result   = self._analyse_triggers(llm_id, user_intent, domain)
        # ? checkpoint

        skill_id = f"skill-{host.now_ms()}"
        skill_store_id, _ = pg_first("svc:skill_store")
        if skill_store_id:
            ask(skill_store_id, "propose_skill", {
                "skill_id":        skill_id,
                "name":            name_result.get("name", "unnamed-skill"),
                "description":     name_result.get("description", ""),
                "procedure":       procedure_result.get("procedure", ""),
                "tags":            trigger_result.get("tags", []),
                "trigger_patterns": trigger_result.get("patterns", []),
            })
        return {"status": "ok", "skill_id": skill_id}

    @signal_handler("cancel")
    def cancel(self) -> None:
        # In-flight extraction can be cancelled without crashing the actor
        host.info("SkillExtraction cancelled")

    @query_handler("status")
    def query_status(self) -> dict:
        return {"task_id": self.task_id, "status": self.status, "progress": self.progress}

@handler("propose_skill")
def propose_skill(self, skill_id: str = "", name: str = "",
                  description: str = "", procedure: str = "",
                  tags: list = None, trigger_patterns: list = None) -> dict:

    # KV: metadata — fast exact-key lookup when the ID is known
    meta = {"skill_id": skill_id, "name": name, "description": description,
            "status": "active", "usage_count": 0,
            "created_at": host.now_ms(), "last_used_at": host.now_ms()}
    host.kv_put(f"skill_meta:{skill_id}", json.dumps(meta))

    # BlobStorage: full procedure text — potentially several paragraphs
    host.blob.upload(f"skill_procedure_{skill_id}", procedure.encode())

    # TupleSpace: keyword indexes — pattern scan at query time, no SQL needed
    for tag in (tags or []):
        host.ts.write(["skill_tag", tag, skill_id, name])
    for pattern in (trigger_patterns or []):
        host.ts.write(["skill_trigger", pattern, skill_id])

    host.incr_counter("skills_created_total", 1)
    return {"status": "ok", "skill_id": skill_id}

@handler("match_skills")
def match_skills(self, query: str = "") -> dict:
    query_words = set(query.lower().split())

    # Scan all trigger entries — None is a wildcard
    all_triggers = host.ts.read_all(["skill_trigger", None, None])

    matched_ids = set()
    for tpl in all_triggers:
        pattern = tpl[1].lower()
        if pattern in query_words or any(w in pattern for w in query_words):
            matched_ids.add(tpl[2])

    skills = []
    for skill_id in matched_ids:
        meta_json = host.kv_get(f"skill_meta:{skill_id}")
        if not meta_json:
            continue
        meta = json.loads(meta_json)
        if meta.get("status") != "active":
            continue
        # Load the full procedure only for matched, active skills
        meta["procedure"] = host.blob.download(f"skill_procedure_{skill_id}").decode()
        skills.append(meta)
        # Track usage for lifecycle decisions
        meta["usage_count"]    += 1
        meta["last_used_at"]   = host.now_ms()
        host.kv_put(f"skill_meta:{skill_id}", json.dumps(meta))

    host.incr_counter("skill_matches_total", len(skills))
    return {"status": "ok", "skills": skills}

@handler("timer_tick", "cast")
def timer_tick(self) -> None:
    now             = host.now_ms()
    thirty_days_ms  = 30 * 24 * 60 * 60 * 1000
    ninety_days_ms  = 90 * 24 * 60 * 60 * 1000

    all_tags = host.ts.read_all(["skill_tag", None, None, None])
    seen     = set()
    for t in all_tags:
        skill_id = t[2]
        if skill_id in seen:
            continue
        seen.add(skill_id)
        meta_json = host.kv_get(f"skill_meta:{skill_id}")
        if not meta_json:
            continue
        meta = json.loads(meta_json)
        age  = now - meta.get("last_used_at", now)
        if meta["status"] == "active" and age > thirty_days_ms:
            meta["status"] = "stale"
            host.kv_put(f"skill_meta:{skill_id}", json.dumps(meta))
        elif meta["status"] == "stale" and age > ninety_days_ms:
            meta["status"] = "archived"
            host.kv_put(f"skill_meta:{skill_id}", json.dumps(meta))

    host.send_after(24 * 60 * 60 * 1000, "timer_tick", {"op": "timer_tick"})

active  --> (30 days unused) -->  stale  --> (90 more days) -->  archived

@actor
class MemoryActor:
    memory_count: int = state(default=0)

    @handler("store_memory")
    def store_memory(self, key: str = "", value: str = "",
                     scope: str = "global", tier: str = "reachable",
                     agent_id: str = "", session_id: str = "") -> dict:
        if not key:
            return {"error": "key required"}
        scoped_key = self._scoped_key(scope, agent_id, session_id, key)

        if tier == "deep":
            # BlobStorage: large, rarely needed, not scanned by default
            host.blob.upload(f"deep_memory_{scoped_key}", value.encode())
        else:
            # KV: durable point lookup
            host.kv_put(scoped_key, str(value))

        # TupleSpace index: queryable by scope and tier regardless of storage layer
        host.ts.write(["memory", scope, tier, key, str(value)[:64]])
        self.memory_count += 1
        return {"status": "ok", "key": key, "scope": scope, "tier": tier}

    @handler("recall_memory")
    def recall_memory(self, key: str = "", scope: str = "global",
                      agent_id: str = "", session_id: str = "") -> dict:
        scoped_key = self._scoped_key(scope, agent_id, session_id, key)
        value = host.kv_get(scoped_key)
        if not value:
            # Try deep tier
            try:
                value = host.blob.download(f"deep_memory_{scoped_key}").decode()
            except Exception:
                pass
        return {"status": "ok", "key": key, "value": value, "found": bool(value)}

    @handler("list_memories")
    def list_memories(self, scope: str = "global", tier: str = None) -> dict:
        pattern = ["memory", scope, tier or None, None, None]
        tuples  = host.ts.read_all(pattern)
        memories = [{"key": t[3], "value": t[4], "tier": t[2]}
                    for t in tuples if len(t) >= 5]
        return {"status": "ok", "memories": memories, "count": len(memories)}

    def _scoped_key(self, scope: str, agent_id: str, session_id: str, key: str) -> str:
        if scope == "agent"   and agent_id:   return f"mem:agent:{agent_id}:{key}"
        if scope == "session" and session_id: return f"mem:session:{session_id}:{key}"
        return f"mem:global:{key}"

// Go — CronSchedulerActor
func (a *CronSchedulerActor) tick() {
    // TryAcquire returns false immediately if another node holds the lock.
    // TTL of 90s exceeds the 60s tick interval, preventing leader gaps.
    acquired, _ := host.Lock().TryAcquire("minihermes", "cron_leader", 90000)
    if !acquired {
        return // another node leads this cycle — nothing to do
    }

    now := host.NowMs()
    for _, jobID := range a.JobIDs {
        job := a.loadJob(jobID)
        if now-job.LastRunAt >= job.IntervalMs {
            payload := map[string]interface{}{
                "job_id": job.JobID, "prompt": job.Prompt, "session_id": job.SessionID,
            }
            // Channel: at-least-once. If agent crashes before ack, job redelivers.
            host.Ch().Send("", "cron:pending", "cron_job", payload)
            job.LastRunAt = now
            a.saveJob(job)
        }
    }
}

@handler("process_cron_job", "cast")
def process_cron_job(self, job_id: str = "", prompt: str = "",
                     session_id: str = "") -> None:
    cron_session = f"cron:{session_id}"

    # Stash the current interactive conversation
    saved_messages = self.messages[:]

    # Load the cron session's own history — completely separate from user sessions
    raw = host.kv_get(f"session_history:{cron_session}")
    self.messages = json.loads(raw) if raw else []

    self._run_agent_loop(prompt, tools=[])

    host.kv_put(f"session_history:{cron_session}", json.dumps(self.messages))
    self.messages = saved_messages  # restore user conversation

    host.send(audit_id, "log_event",
              {"event_type": "cron_executed", "detail": f"job_id={job_id}"})

curl -X POST http://localhost:8091/api/v1/actors/cron_scheduler/create_job \
  -d '{
    "job_id":     "daily-digest",
    "prompt":     "Summarize today'\''s tasks and send a digest email",
    "schedule":   "every_24h",
    "session_id": "cron-digest"
  }'

@handler("check_and_compress")
def check_and_compress(self, messages: list = None, token_budget: int = 4096) -> dict:
    messages        = messages or []
    estimated_tokens = sum(len(str(m)) // 4 for m in messages)

    if estimated_tokens < token_budget * 0.75:
        return {"compressed": False, "messages": messages}

    system_msgs  = [m for m in messages if m.get("role") == "system"]
    other_msgs   = [m for m in messages if m.get("role") != "system"]
    recent_count = max(4, len(other_msgs) // 3)
    middle       = other_msgs[:-recent_count]
    recent       = other_msgs[-recent_count:]

    if len(middle) < 2:
        return {"compressed": False, "messages": messages}

    # Archive the full original before compression — preserves audit trail
    if self.session_id:
        host.kv_put(f"full_history_archive:{self.session_id}", json.dumps(messages))

    llm_id, _ = pg_first("svc:llm_gateway")
    summary_resp = ask(llm_id, "completion", {
        "messages": [
            {"role": "system",
             "content": "Summarize this conversation history concisely. "
                        "Preserve key facts, tool results, and decisions."},
            {"role": "user", "content": json.dumps(middle)}
        ],
        "tools": []
    })

    summary_text = summary_resp.get("response", {}).get("content", "")
    summary_msg  = {"role": "assistant",
                    "content": f"[Conversation summary: {summary_text}]",
                    "is_summary": True}

    compressed = system_msgs + [summary_msg] + recent
    host.incr_counter("context_compressions_total", 1)
    return {"compressed": True, "messages": compressed,
            "original_count": len(messages), "compressed_count": len(compressed)}

@fsm_actor(states=["allow", "review", "approved", "denied"], initial="allow")
class GuardrailsGateActor:
    # tool_name ? "allow" | "deny" | "review"
    policies: dict = state(default_factory=dict)
    deny_count: int = state(default=0)

    @handler("check_tool")
    def check_tool(self, tool_name: str = "", input: dict = None) -> dict:
        policy = self.policies.get(tool_name, "allow")

        if policy == "deny":
            self.deny_count += 1
            host.incr_counter("tool_denials_total", 1)
            host.send(audit_id, "log_event",
                      {"event_type": "tool_denied", "detail": f"tool={tool_name}"})
            return {"decision": "deny", "reason": f"{tool_name} is blocked by policy"}

        if policy == "review":
            # FSM transitions to review — observable by operators via get_state
            self.fsm_state = "review"
            host.send(audit_id, "log_event",
                      {"event_type": "tool_review", "detail": f"tool={tool_name}"})
            # Production: pause here and await human approval via Channel
            self.fsm_state = "approved"
            return {"decision": "allow", "reviewed": True}

        return {"decision": "allow"}

    @handler("set_policy")
    def set_policy(self, tool_name: str = "", decision: str = "allow") -> dict:
        self.policies[tool_name] = decision
        host.send(audit_id, "log_event",
                  {"event_type": "policy_set",
                   "detail": f"tool={tool_name} decision={decision}"})
        return {"status": "ok", "tool_name": tool_name, "decision": decision}

    @handler("get_state")
    def get_state(self) -> dict:
        return {"fsm_state": self.fsm_state, "policies": self.policies,
                "deny_count": self.deny_count}

# Block a dangerous tool immediately — affects all in-flight and future calls
curl -X POST http://localhost:8091/api/v1/actors/guardrails/set_policy \
  -d '{"tool_name":"delete_file","decision":"deny"}'

# Route a sensitive tool through human review
curl -X POST http://localhost:8091/api/v1/actors/guardrails/set_policy \
  -d '{"tool_name":"send_email","decision":"review"}'

# Register a weather API as a tool — takes effect immediately
curl -X POST http://localhost:8091/api/v1/actors/tool_executor/register_tool \
  -d '{
    "name":        "weather",
    "description": "Get current weather for a city",
    "input_schema": {"type":"object","properties":{"city":{"type":"string"}}},
    "handler_type": "service_link",
    "handler_config": {"link_name":"openweather","path":"/data/2.5/weather","method":"GET"}
  }'

@actor
class ToolExecutorActor:
    tools: dict     = state(default_factory=dict)   # name ? spec
    exec_count: int = state(default=0)

    @init_handler
    def on_init(self, config: dict) -> None:
        self.tools = {t["name"]: t for t in _BUILTIN_TOOLS}
        host.process_groups.join("svc:tool_executor")

    @handler("register_tool")
    def register_tool(self, name: str = "", description: str = "",
                      input_schema: dict = None, handler_type: str = "builtin",
                      handler_config: dict = None) -> dict:
        self.tools[name] = {
            "name": name, "description": description,
            "input_schema": input_schema or {},
            "handler_type": handler_type,
            "handler_config": handler_config or {}
        }
        return {"status": "ok", "name": name}

    @handler("execute")
    def execute(self, name: str = "", input: dict = None) -> dict:
        input = input or {}
        if name not in self.tools:
            return {"error": f"unknown tool: {name}"}
        self.exec_count += 1
        host.incr_counter(f"tool_{name}_total", 1)

        spec = self.tools[name]
        if spec.get("handler_type") == "service_link":
            cfg  = spec.get("handler_config", {})
            resp = host.http_fetch(cfg["link_name"], cfg.get("method","GET"),
                                   cfg["path"], input)
            return {"result": resp}

        # Built-in handlers
        if name == "calculator":
            expr = input.get("expression", "0")
            try:
                result = eval(expr, {"__builtins__": {}})  # demo only — see gaps section
                return {"result": str(result)}
            except Exception as e:
                return {"error": str(e)}
        if name == "memory_store":
            mem_id, _ = pg_first("svc:memory")
            if mem_id:
                return ask(mem_id, "store_memory", input) or {}
        if name == "memory_recall":
            mem_id, _ = pg_first("svc:memory")
            if mem_id:
                return ask(mem_id, "recall_memory", input) or {}

        return {"result": f"[simulated] {name} executed"}

# Every actor announces itself on startup
host.process_groups.join("svc:agent")

# Callers find the first available member — location-transparent
agent_id, err = pg_first("svc:agent")
result = ask(agent_id, "chat", {"message": "Hello"})

// Go version — same pattern
agentID, err := host.PG().First("svc:agent")

# On startup — declare what this actor can do
host.registry.register(ctx="", object_type="actor",
                        object_id=self.actor_id,
                        object_category="skill_store",
                        capabilities=["match_skills", "propose_skill", "lifecycle"])

# Caller — find an actor that specifically supports skill matching
actors = host.registry.discover(ctx="", object_type="actor",
                                 object_category="skill_store",
                                 required_capability="match_skills")
skill_id = actors[0]["object_id"] if actors else None

// Go — capability-aware lookup
agentID, err := registryFirst("agent", "svc:agent", "tool_use")

@event_actor
class AuditEventActor:
    event_count: int = state(default=0)

    @init_handler
    def on_init(self, config: dict) -> None:
        host.process_groups.join("svc:audit")

    @handler("log_event", "cast")  # "cast" = fire-and-forget, no reply
    def log_event(self, event_type: str = "", detail: str = "",
                  timestamp: int = 0) -> None:
        ts = timestamp or host.now_ms()
        host.ts.write(["audit", event_type, ts, detail])
        self.event_count += 1

    @handler("query_events")
    def query_events(self, event_type: str = None) -> dict:
        pattern = ["audit", event_type or None, None, None]
        events  = host.ts.read_all(pattern)
        return {"status": "ok", "events": events, "count": len(events)}

_SERVICE_GROUPS = [
    "svc:llm_gateway", "svc:tool_executor", "svc:agent",
    "svc:skill_store", "svc:guardrails", "svc:audit",
    "svc:cron_scheduler", "svc:session_manager", "svc:memory",
    "svc:context_compressor", "svc:health_monitor",
]

@actor
class HealthMonitorActor:
    poll_count:      int  = state(default=0)
    last_poll_ms:    int  = state(default=0)
    group_health:    dict = state(default_factory=dict)
    poll_interval_ms: int = state(default=5000)

    @init_handler
    def on_init(self, config: dict) -> None:
        host.process_groups.join("svc:health_monitor")
        host.send_after(self.poll_interval_ms, "poll_tick", {"op": "poll_tick"})

    @handler("poll_tick", "cast")
    def poll_tick(self) -> None:
        health = {}
        for grp in _SERVICE_GROUPS:
            try:
                members      = host.process_groups.members(grp)
                health[grp]  = len(members)
            except Exception:
                health[grp] = 0

        self.group_health  = health
        self.poll_count   += 1
        self.last_poll_ms  = host.now_ms()

        host.ts.write(["health_snapshot", self.last_poll_ms, json.dumps(health)])
        # Each tick reschedules the next — no external scheduler
        host.send_after(self.poll_interval_ms, "poll_tick", {"op": "poll_tick"})

    @handler("get_health")
    def get_health(self) -> dict:
        degraded = [g for g, c in self.group_health.items() if c == 0]
        return {
            "status":       "ok" if not degraded else "degraded",
            "group_health": self.group_health,
            "healthy":      len(self.group_health) - len(degraded),
            "degraded":     degraded,
        }

brew tap tinygo-org/tools && brew install tinygo
cargo install wasm-tools
npm install -g @bytecodealliance/jco

pip install -e path/to/sdks/python

brew install ollama
ollama run llama3.2   # pulls ~2GB on first run

# Python
cd examples/python/apps/minihermes
./build.sh                       # componentize-py ? WASM Component Model binary
pytest test_minihermes.py -v     # unit tests, no live node required

# Go
cd examples/go/apps/minihermes
./build.sh                       # TinyGo ? wasm-tools ? component binary
go test ./... -v                 # unit tests, no live node required

# Start a PlexSpaces node first — see docs/getting-started.md
cd examples/go/apps/minihermes
./test.sh 8091                   # 21 steps, roughly 2 minutes

# Basic agent chat
ask "agent" '{"op":"chat","message":"Hello","session_id":"test-1"}'

# Tool use — triggers guardrail check before execution
ask "agent" '{"op":"chat","message":"Calculate 42 * 17","session_id":"test-1"}'

# Hot-swap LLM provider
ask "llm_gateway" '{"op":"switch_provider","provider":"anthropic","model":"claude-opus-4-8"}'

# Register a new tool at runtime
ask "tool_executor" '{
  "op":"register_tool","name":"weather",
  "description":"Get weather for a city",
  "input_schema":{"type":"object","properties":{"city":{"type":"string"}}},
  "handler_type":"service_link",
  "handler_config":{"link_name":"openweather","path":"/data/2.5/weather","method":"GET"}
}'

# Create a cron job
ask "cron_scheduler" '{
  "op":"create_job","job_id":"morning-digest",
  "prompt":"Summarize pending tasks","schedule":"every_24h","session_id":"cron-main"
}'

# Block a tool via guardrails
ask "guardrails" '{"op":"set_policy","tool_name":"delete_file","decision":"deny"}'

# Query health across all service groups
ask "health_monitor" '{"op":"get_health"}'

# Query audit trail for tool executions
ask "audit_event" '{"op":"query_events","event_type":"tool_executed"}'

ai-gh-issue-picker.yaml   (~100 lines)  — polls GitHub, submits jobs
ai-gh-implement.yaml      (~500 lines)  — plan -> implement -> test -> verify -> PR -> monitor -> learn
ai-gh-cleanup.yaml        (~80 lines)   — stale workspace and branch cleanup

job_type: ai-gh-issue-picker
cron_trigger: "0 * * * * * *"   # every minute (7-field cron)
max_concurrency: 1               # only one picker at a time

skip_if: >-
  {{if ge (CountByJobTypeAndState "ai-gh-implement" "PENDING") 10}} true {{end}}

import json, os, subprocess

repo = f"{os.environ['GH_ORG']}/{os.environ['GH_REPO']}"

def gh(*args):
    r = subprocess.run(["gh"] + list(args), capture_output=True, text=True)
    return r

r = gh("issue", "list", "-R", repo,
       "--label", os.environ["PICKUP_LABEL"], "--state", "open",
       "--limit", os.environ.get("MAX_PENDING", "10"),
       "--json", "number,title,url")
issues = json.loads(r.stdout) if r.returncode == 0 else []

for issue in issues:
    gh("issue", "edit", str(issue["number"]), "-R", repo,
       "--remove-label", os.environ["PICKUP_LABEL"],
       "--add-label", os.environ["INPROGRESS_LABEL"])

issues_json = json.dumps(issues, separators=(',', ':'))
with open("issues.json", "w") as f:
    f.write(issues_json + "\n")
print(f"::set-output name=IssuesJSON::{issues_json}")

environment:
  SUBMITTED_IDS: >-
    {{if .IssuesJSON}}{{SubmitJobsFromJSON "ai-gh-implement" .IssuesJSON
        (printf "GitHubOrg=%s" .GitHubOrg) (printf "GitHubRepo=%s" .GitHubRepo)}}{{end}}
  PENDING_COUNT: '{{CountByJobTypeAndState "ai-gh-implement" "PENDING"}}'

- task_type: implement
  on_exit_code:
    COMPLETED: unit-test
    "2": notify-blocked    # model signals blocked
    "3": fix-tests         # tests failing
  on_failed: notify-blocked

- task_type: plan
  timeout: 15m

- task_type: implement
  timeout: 45m

- task_type: unit-test
  timeout: 10m

- task_type: self-verify
  timeout: 15m

- task_type: cleanup
  always_run: true    # runs even if the job fails
  timeout: 1m

- task_type: self-verify
  dependencies:
    - setup       # downloads meta.env
    - implement   # downloads impl_result.json, impl_conversation.txt, impl_diff.patch
  script:
    - |
      TASK_DIR="$PWD"          # capture executor dir before any cd
      source "$TASK_DIR/meta.env"
      cd "$WS/repo"
      # all artifacts available in $TASK_DIR/

TASK_DIR="$PWD"
source "$TASK_DIR/meta.env"
cd "$WS/repo"
# ... do work ...
jq ... > "$TASK_DIR/result.json"   # write to TASK_DIR, not to repo

NONCE=$(head -c 4 /dev/urandom | xxd -p)
BRANCH="ai/{{.IssueNumber}}-${SLUG}-${NONCE}"
# e.g., ai/42-fix-login-timeout-a3f1

claude --print --dangerously-skip-permissions --model "$MODEL" --max-turns 100 \
  "$(cat /tmp/impl_prompt.txt)" 2>&1 | tee "$TASK_DIR/impl_conversation.txt"

# Extract the last JSON object with a "status" key
STATUS_JSON=$(grep -oE '\{[^{}]*"status"[^{}]*\}' \
  "$TASK_DIR/impl_conversation.txt" | tail -1)
STATUS=$(echo "$STATUS_JSON" | jq -r '.status // "UNKNOWN"')
[ "$STATUS" = "BLOCKED" ] && exit 2
[ "$STATUS" = "TESTS_FAILING" ] && exit 3

claude --print --model opus --max-turns 30 \
  "Use the ygs-wbs skill approach:
   1. Explore the codebase
   2. Decompose into vertical-slice tasks
   3. Write PLANS/{issue-slug}-{number}-plan.md with acceptance criteria"

{
  "job_type": "ai-gh-implement",
  "description": "#42: Fix login timeout | [org/repo](https://github.com/org/repo)",
  "params": {
    "IssueLink": "[#42: Fix login timeout](https://github.com/org/repo/issues/42)",
    "BranchLink": "[ai/42-fix-login-a3f1](https://github.com/org/repo/tree/ai/42-fix-login-a3f1)",
    "PRLink": ""
  }
}

- task_type: monitor-pr
  method: SHELL
  timeout: 24h

- task_type: learn
  method: SHELL
  # reviews PR feedback, writes to ~/claude_workspace/learn_context/

container:
  image: ghcr.io/formicary-ai/agent-worker:latest
  env_from:
    - secret_ref: claude-bedrock-settings
    - secret_ref: ai-agent-secrets

container:
  env_value_from:
    - name: ANTHROPIC_API_KEY
      secret_name: ai-agent-secrets
      key: anthropic-api-key

container:
  service_account: ai-agent-irsa-sa

skip_if: >-
  {{if ge (CountByJobTypeAndState "ai-gh-implement" "PENDING" "EXECUTING") 10}} true {{end}}

# 1. Prerequisites (one-time)
npm install -g @anthropic-ai/claude-code
gh auth login

# 2. Start Formicary (queen + embedded ant worker)
docker pull plexobject/formicary
docker run plexobject/formicary

# 3. Deploy workflow definitions
git clone https://github.com/bhatti/formicary.git
cd docs/examples
./deploy-ai-workflows.sh --mode shell --repo your-org/your-repo --setup-labels

# 4. Set org config so the picker knows where to look
curl -X POST http://localhost:7777/api/orgs/default/configs \
  -H 'Content-Type: application/json' \
  -d '{"name":"GitHubOrg","value":"your-org"}'
curl -X POST http://localhost:7777/api/orgs/default/configs \
  -H 'Content-Type: application/json' \
  -d '{"name":"GitHubRepo","value":"your-repo"}'

# 5. Label an issue — the picker fires within 1 minute
gh issue edit 1 --repo your-org/your-repo --add-label "ai-ready"

# 6. Watch it run
open http://localhost:7777

TAILSCALE_IP=$(python3 -c "import socket; print(socket.gethostbyname('ai'))")

kubectl create namespace formicary-ai

kubectl create secret generic claude-bedrock-settings \
  --namespace=formicary-ai \
  --from-literal=ANTHROPIC_BEDROCK_BASE_URL=http://${TAILSCALE_IP}/bedrock \
  --from-literal=CLAUDE_CODE_USE_BEDROCK=1 \
  --from-literal=CLAUDE_CODE_SKIP_BEDROCK_AUTH=1 \
  --from-literal=ANTHROPIC_DEFAULT_OPUS_MODEL=us.anthropic.claude-opus-4-6-v1 \
  --from-literal=ANTHROPIC_DEFAULT_SONNET_MODEL=us.anthropic.claude-sonnet-4-6 \
  --from-literal=ANTHROPIC_DEFAULT_HAIKU_MODEL=us.anthropic.claude-haiku-4-5-20251001-v1:0

kubectl create secret generic ai-agent-secrets \
  --namespace=formicary-ai \
  --from-literal=github-token=$(gh auth token)

kubectl create secret generic ai-agent-secrets \
  --from-literal=anthropic-api-key=sk-ant-... \
  --from-literal=github-token=$(gh auth token)

# Create instance with L4 GPU
gcloud compute instances create vllm-test \
  --zone=us-central1-a \
  --machine-type=g2-standard-8 \
  --accelerator=type=nvidia-l4,count=1 \
  --image-family=ubuntu-2004-lts \
  --image-project=ubuntu-os-cloud \
  --boot-disk-size=200GB \
  --boot-disk-type=pd-ssd \
  --maintenance-policy=TERMINATE

# SSH into instance
gcloud compute ssh vllm-test --zone=us-central1-a

# Update system
sudo apt update && sudo apt upgrade -y

# Install CUDA 11.8
wget https://developer.download.nvidia.com/compute/cuda/11.8.0/local_installers/cuda_11.8.0_520.61.05_linux.run
sudo sh cuda_11.8.0_520.61.05_linux.run --silent --toolkit

# Add to PATH
echo 'export PATH=/usr/local/cuda-11.8/bin:$PATH' >> ~/.bashrc
echo 'export LD_LIBRARY_PATH=/usr/local/cuda-11.8/lib64:$LD_LIBRARY_PATH' >> ~/.bashrc
source ~/.bashrc

# Verify
nvidia-smi  # Should show your L4 GPU
nvcc --version  # Should show CUDA 11.8

# Install Python 3.10
sudo apt install -y python3.10 python3.10-venv python3-pip

# Clone the repository
git clone https://github.com/bhatti/vllm-tutorial.git
cd vllm-tutorial

# Create virtual environment
python3 -m venv venv
source venv/bin/activate

# Install dependencies
pip install --upgrade pip
pip install -r requirements.txt

# Test vLLM installation
python -c "import vllm; print(f'vLLM version: {vllm.__version__}')"

# Quick functionality test
python examples/01_basic_vllm.py

Loading model microsoft/phi-2...
Model loaded in 8.3 seconds

Generating response...
Generated 50 tokens in 987ms
Throughput: 41.5 tokens/sec

? vLLM is working!

   git clone https://github.com/bhatti/vllm-tutorial.git
   cd vllm-tutorial

   # Follow setup instructions in README
   python examples/01_basic_vllm.py

   # See the actual results from GCP L4 testing
   cat benchmarks/results/01_throughput_results.json

def classify_complexity(self, prompt: str) -> str:
    """
    Classify prompt complexity to select appropriate model

    Rules:
    - Simple: Definitions, quick facts, <50 words
    - Medium: Summaries, comparisons, 50-150 words
    - Complex: Deep analysis, multi-step reasoning, >150 words
    """
    word_count = len(prompt.split())

    # Keywords indicating complexity
    complex_keywords = [
        "analyze", "compare", "evaluate", "assess risk",
        "recommend", "predict", "forecast", "implications"
    ]

    medium_keywords = [
        "summarize", "explain", "describe", "list",
        "what are", "how does", "differences"
    ]

    has_complex = any(kw in prompt.lower() for kw in complex_keywords)
    has_medium = any(kw in prompt.lower() for kw in medium_keywords)

    # Classification logic
    if word_count > 150 or has_complex:
        return "complex"
    elif word_count > 50 or has_medium:
        return "medium"
    else:
        return "simple"

Query: "What is EBITDA?"
Classified as: simple ? Routed to: Phi-2
Cost: $0.00002038
Latency: 4,843ms (first request, includes model loading)
Quality: ? Perfect (simple definition)

Query: "Summarize Apple's Q4 2024 earnings highlights"
Classified as: medium ? Routed to: Mistral-7B
Cost: $0.00000865
Latency: 4,827ms
Quality: ? Good summary

Query: "Analyze Microsoft's 10-K risk factors and assess their potential impact on future earnings"
Classified as: complex ? Routed to: Llama-3-8B
Cost: $0.00001382
Latency: 4,836ms
Quality: ? Detailed analysis

from typing import Dict, Optional
from dataclasses import dataclass
from vllm import LLM, SamplingParams

@dataclass
class ModelConfig:
    """Configuration for a model tier"""
    name: str
    complexity: str  # "simple", "medium", "complex"
    cost_per_1k_tokens: float
    max_tokens: int

class IntelligentRouter:
    """
    Production-ready intelligent router with:
    - Complexity-based routing
    - Budget enforcement
    - Cost tracking
    - Fallback handling
    """

    def __init__(self, daily_budget_usd: float = 100.0):
        self.daily_budget_usd = daily_budget_usd
        self.total_cost_today = 0.0

        # Model configurations
        self.models = {
            "phi-2": ModelConfig(
                name="microsoft/phi-2",
                complexity="simple",
                cost_per_1k_tokens=0.0001,
                max_tokens=1024,
            ),
            "mistral-7b": ModelConfig(
                name="mistralai/Mistral-7B-Instruct-v0.2",
                complexity="medium",
                cost_per_1k_tokens=0.0002,
                max_tokens=2048,
            ),
            "llama-3-8b": ModelConfig(
                name="meta-llama/Meta-Llama-3-8B",
                complexity="complex",
                cost_per_1k_tokens=0.0003,
                max_tokens=4096,
            ),
        }

        # Initialize LLM (in production, these would be separate instances)
        self.llm = LLM(
            model=self.models["phi-2"].name,
            trust_remote_code=True,
            gpu_memory_utilization=0.9,
        )

    def route_request(self, prompt: str, max_tokens: int = 200) -> Dict:
        """
        Route request to appropriate model

        Returns:
            Dict with 'response', 'model_used', 'cost', 'latency_ms'
        """
        # Step 1: Classify complexity
        complexity = self.classify_complexity(prompt)

        # Step 2: Select model
        model_id = self._select_model(complexity)
        model_config = self.models[model_id]

        # Step 3: Check budget
        estimated_cost = self._estimate_cost(model_config, prompt, max_tokens)
        if self.total_cost_today + estimated_cost > self.daily_budget_usd:
            # Budget exceeded - fallback to cheapest model
            model_id = "phi-2"
            model_config = self.models[model_id]

        # Step 4: Generate response
        sampling_params = SamplingParams(
            temperature=0.7,
            top_p=0.9,
            max_tokens=max_tokens,
        )

        start_time = time.time()
        outputs = self.llm.generate([prompt], sampling_params)
        latency_ms = (time.time() - start_time) * 1000

        # Step 5: Track cost
        tokens_generated = len(outputs[0].outputs[0].token_ids)
        actual_cost = self._calculate_cost(model_config, prompt, tokens_generated)
        self.total_cost_today += actual_cost

        return {
            "response": outputs[0].outputs[0].text,
            "model_used": model_id,
            "cost_usd": actual_cost,
            "latency_ms": latency_ms,
            "tokens_generated": tokens_generated,
        }

    def _select_model(self, complexity: str) -> str:
        """Select model based on complexity"""
        for model_id, config in self.models.items():
            if config.complexity == complexity:
                return model_id
        return "phi-2"  # Default fallback

    def _estimate_cost(self, config: ModelConfig, prompt: str, max_tokens: int) -> float:
        """Estimate cost before generation"""
        input_tokens = len(prompt) / 4  # Rough estimate
        total_tokens = input_tokens + max_tokens
        return (total_tokens / 1000) * config.cost_per_1k_tokens

    def _calculate_cost(self, config: ModelConfig, prompt: str, tokens_generated: int) -> float:
        """Calculate actual cost after generation"""
        input_tokens = len(prompt) / 4
        total_tokens = input_tokens + tokens_generated
        return (total_tokens / 1000) * config.cost_per_1k_tokens

# Initialize router with daily budget
router = IntelligentRouter(daily_budget_usd=100.0)

# Route a simple query
result = router.route_request("What is gross margin?")
print(f"Model used: {result['model_used']}")  # phi-2
print(f"Cost: ${result['cost_usd']:.6f}")     # $0.000020

# Route a complex query
result = router.route_request(
    "Analyze Tesla's competitive positioning in the EV market "
    "and provide investment recommendations based on recent trends"
)
print(f"Model used: {result['model_used']}")  # llama-3-8b
print(f"Cost: ${result['cost_usd']:.6f}")     # $0.000138

from dataclasses import dataclass
from datetime import datetime

@dataclass
class BudgetConfig:
    """Budget configuration with multiple enforcement levels"""
    max_cost_per_request: float = 0.50        # Level 1: prevent accidents
    daily_budget_usd: float = 100.0           # Level 2: daily cap
    monthly_budget_usd: float = 3000.0        # Level 3: monthly cap
    warning_threshold_pct: float = 0.80       # Warn at 80%

class BudgetEnforcer:
    """Hard budget enforcement - prevents spending, not just monitors"""
    
    def __init__(self, config: BudgetConfig):
        self.config = config
        self.daily_spend = 0.0
        self.monthly_spend = 0.0
        # ... implementation
    
    def check_budget(self, estimated_cost: float) -> Dict:
        """Check BEFORE generating - this is the key difference"""
        
        # Level 1: Per-request limit
        if estimated_cost > self.config.max_cost_per_request:
            return {"action": "reject", "reason": "Request too expensive"}
        
        # Level 2: Daily budget
        if self.daily_spend + estimated_cost > self.config.daily_budget_usd:
            return {"action": "downgrade", "reason": "Daily limit approaching"}
        
        # Level 3: Monthly budget
        if self.monthly_spend + estimated_cost > self.config.monthly_budget_usd:
            return {"action": "downgrade", "reason": "Monthly limit approaching"}
        
        return {"action": "allow"}

[Fixed System Prompt - 500 tokens]
You are a financial analyst AI assistant specializing in:
- Earnings report analysis
- SEC filing interpretation
- Market sentiment analysis
...

[User Query - 50 tokens]
What is EBITDA?

[Response - 100 tokens]
EBITDA stands for...

from vllm import LLM

# WITHOUT prefix caching
llm = LLM(
    model="microsoft/phi-2",
    trust_remote_code=True,
)

# WITH prefix caching (80% cost reduction!)
llm = LLM(
    model="microsoft/phi-2",
    trust_remote_code=True,
    enable_prefix_caching=True,  # <-- That's it!
)

Request 1: $0.00010188 (full cost)
Request 2: $0.00010188 (full cost)
Request 3: $0.00010188 (full cost)
...
Total: $0.00112068 (11 × $0.00010188)

Request 1: $0.00002038 (full cost - establishes cache)
Request 2: $0.00000414 (80% cheaper - uses cache!)
Request 3: $0.00000409 (80% cheaper)
Request 4: $0.00000865 (80% cheaper)
...
Total: $0.00010031

Savings: $0.00102037 (91% reduction!)
Cache hit rate: 90.9% (10/11 requests)

# Benchmark code
class QuantizationBenchmark:
    def benchmark_quantization(self, quantization: str):
        """
        Test quantization scheme

        Args:
            quantization: "none" (FP16), "fp8", "awq", or "gptq"
        """
        llm_kwargs = {
            "model": "microsoft/phi-2",
            "trust_remote_code": True,
            "gpu_memory_utilization": 0.9,
            "max_model_len": 1024,
        }

        # Add quantization if specified
        if quantization != "none":
            llm_kwargs["quantization"] = quantization

        # Load model
        start = time.time()
        llm = LLM(**llm_kwargs)
        load_time = time.time() - start

        # Measure memory
        gpu_memory = torch.cuda.memory_allocated() / 1e9  # GB

        # Benchmark generation
        prompt = "Explain quantum computing in simple terms"
        sampling_params = SamplingParams(max_tokens=100)

        start = time.time()
        outputs = llm.generate([prompt], sampling_params)
        latency_ms = (time.time() - start) * 1000

        return {
            "quantization": quantization,
            "memory_gb": gpu_memory,
            "load_time_sec": load_time,
            "latency_ms": latency_ms,
            "tokens_per_sec": 100 / (latency_ms / 1000),
        }

from vllm import LLM

# Option 1: Use pre-quantized AWQ model
llm = LLM(
    model="TheBloke/Mistral-7B-Instruct-v0.2-AWQ",  # Pre-quantized!
    quantization="awq",
    trust_remote_code=True,
    gpu_memory_utilization=0.9,
)

# That's it! 3.7x smaller, ready to use

# Mistral-7B in FP16
llm_fp16 = LLM(model="mistralai/Mistral-7B-Instruct-v0.2")
# Memory: ~16GB VRAM
# Fits on: A100-40GB, L4 (barely)

# Mistral-7B in AWQ
llm_awq = LLM(
    model="TheBloke/Mistral-7B-Instruct-v0.2-AWQ",
    quantization="awq"
)
# Memory: ~4.3GB VRAM
# Fits on: T4-16GB, L4 (comfortably), even RTX 3090

# Savings: 72% memory reduction!

from prometheus_client import Counter, Histogram, Gauge, generate_latest
from typing import Dict
import time

class VLLMMetrics:
    """
    Production metrics for vLLM serving

    Tracks:
    - Request counts (total, success, failure)
    - Latency distributions (P50, P95, P99)
    - Token throughput
    - Cost tracking
    - Model distribution
    """

    def __init__(self):
        # Request counters
        self.requests_total = Counter(
            'vllm_requests_total',
            'Total number of requests',
            ['model', 'status']
        )

        # Latency histogram
        self.latency = Histogram(
            'vllm_latency_ms',
            'Request latency in milliseconds',
            ['model'],
            buckets=[10, 50, 100, 250, 500, 1000, 2500, 5000, 10000]
        )

        # Token metrics
        self.tokens_generated = Counter(
            'vllm_tokens_generated_total',
            'Total tokens generated',
            ['model']
        )

        self.tokens_per_second = Gauge(
            'vllm_tokens_per_second',
            'Current tokens per second',
            ['model']
        )

        # Cost tracking
        self.cost_usd = Counter(
            'vllm_cost_usd_total',
            'Total cost in USD',
            ['model']
        )

        self.daily_cost = Gauge(
            'vllm_daily_cost_usd',
            'Cost today in USD'
        )

        # GPU memory
        self.gpu_memory_used = Gauge(
            'vllm_gpu_memory_used_gb',
            'GPU memory used in GB'
        )

        self.gpu_memory_total = Gauge(
            'vllm_gpu_memory_total_gb',
            'Total GPU memory in GB'
        )

        # Cache metrics
        self.cache_hit_rate = Gauge(
            'vllm_cache_hit_rate',
            'Prefix cache hit rate'
        )

    def record_request(
        self,
        model: str,
        latency_ms: float,
        tokens: int,
        cost_usd: float,
        success: bool,
        cached: bool = False
    ):
        """Record request metrics"""

        # Update counters
        status = "success" if success else "failure"
        self.requests_total.labels(model=model, status=status).inc()

        if success:
            # Latency
            self.latency.labels(model=model).observe(latency_ms)

            # Tokens
            self.tokens_generated.labels(model=model).inc(tokens)
            tokens_per_sec = tokens / (latency_ms / 1000)
            self.tokens_per_second.labels(model=model).set(tokens_per_sec)

            # Cost
            self.cost_usd.labels(model=model).inc(cost_usd)

    def update_gpu_memory(self):
        """Update GPU memory metrics"""
        if torch.cuda.is_available():
            used_gb = torch.cuda.memory_allocated() / 1e9
            total_gb = torch.cuda.get_device_properties(0).total_memory / 1e9

            self.gpu_memory_used.set(used_gb)
            self.gpu_memory_total.set(total_gb)

    def export_metrics(self) -> str:
        """Export Prometheus metrics"""
        return generate_latest().decode('utf-8')

# Usage in FastAPI
from fastapi import FastAPI

app = FastAPI()
metrics = VLLMMetrics()

@app.post("/generate")
async def generate(request: GenerateRequest):
    start = time.time()

    try:
        # Generate response
        result = llm.generate(request.prompt)

        # Record success metrics
        latency_ms = (time.time() - start) * 1000
        metrics.record_request(
            model="phi-2",
            latency_ms=latency_ms,
            tokens=len(result.tokens),
            cost_usd=calculate_cost(result),
            success=True,
        )

        return result

    except Exception as e:
        # Record failure
        latency_ms = (time.time() - start) * 1000
        metrics.record_request(
            model="phi-2",
            latency_ms=latency_ms,
            tokens=0,
            cost_usd=0,
            success=False,
        )
        raise

@app.get("/metrics")
async def get_metrics():
    """Prometheus scrape endpoint"""
    metrics.update_gpu_memory()
    return Response(
        content=metrics.export_metrics(),
        media_type="text/plain"
    )

from langfuse import Langfuse
import os

# Initialize Langfuse
langfuse = Langfuse(
    public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
    secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
    host=os.getenv("LANGFUSE_HOST", "http://localhost:3001"),
)

def generate_with_observability(prompt: str, user_id: str, metadata: Dict = None):
    """Generate response with full Langfuse tracing"""

    # Create trace
    trace = langfuse.trace(
        name="financial_analysis",
        user_id=user_id,
        metadata=metadata or {},
    )

    # Start generation span
    generation = trace.generation(
        name="vllm_generate",
        model="microsoft/phi-2",
        input=prompt,
        metadata={
            "quantization": "awq",
            "max_tokens": 200,
        }
    )

    # Generate
    start = time.time()
    result = llm.generate(prompt)
    latency_ms = (time.time() - start) * 1000

    # Calculate cost
    tokens_in = len(prompt) / 4
    tokens_out = len(result.tokens)
    cost_usd = ((tokens_in + tokens_out) / 1000) * 0.0001

    # End span with metrics
    generation.end(
        output=result.text,
        usage={
            "input_tokens": int(tokens_in),
            "output_tokens": tokens_out,
            "total_tokens": int(tokens_in + tokens_out),
        },
        metadata={
            "latency_ms": latency_ms,
            "cost_usd": cost_usd,
            "model": "phi-2",
        }
    )

    return result

# Usage
result = generate_with_observability(
    prompt="Analyze Apple's Q4 earnings",
    user_id="analyst_001",
    metadata={
        "team": "equity_research",
        "department": "finance",
    }
)

from typing import Callable
from dataclasses import dataclass
import time

@dataclass
class RetryConfig:
    """Retry configuration"""
    max_retries: int = 3
    initial_delay: float = 1.0
    max_delay: float = 60.0
    exponential_base: float = 2.0

def retry_with_backoff(config: RetryConfig = RetryConfig()):
    """
    Decorator: Retry with exponential backoff

    Example:
        @retry_with_backoff()
        def generate_text(prompt):
            return llm.generate(prompt)
    """
    def decorator(func: Callable) -> Callable:
        def wrapper(*args, **kwargs):
            delay = config.initial_delay

            for attempt in range(config.max_retries):
                try:
                    return func(*args, **kwargs)

                except Exception as e:
                    if attempt == config.max_retries - 1:
                        raise  # Last attempt, re-raise

                    error_type = classify_error(e)

                    # Don't retry on invalid input
                    if error_type == ErrorType.INVALID_INPUT:
                        raise

                    print(f"??  Attempt {attempt + 1} failed: {error_type.value}")
                    print(f"   Retrying in {delay:.1f}s...")
                    time.sleep(delay)

                    # Exponential backoff
                    delay = min(delay * config.exponential_base, config.max_delay)

            raise RuntimeError(f"Failed after {config.max_retries} retries")

        return wrapper
    return decorator

# Usage
@retry_with_backoff(RetryConfig(max_retries=3, initial_delay=1.0))
def generate_with_retry(prompt: str):
    """Generate with automatic retry on failure"""
    return llm.generate(prompt)

# This will retry up to 3 times with exponential backoff
result = generate_with_retry("Analyze earnings report")

from datetime import datetime, timedelta
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing recovery

class CircuitBreaker:
    """
    Circuit breaker for fault tolerance

    Prevents cascading failures by stopping calls to
    failing services
    """

    def __init__(
        self,
        failure_threshold: int = 5,
        timeout: int = 60,
        expected_exception: type = Exception
    ):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.expected_exception = expected_exception

        self.failure_count = 0
        self.last_failure_time = None
        self.state = CircuitState.CLOSED

    def call(self, func: Callable, *args, **kwargs):
        """Execute function with circuit breaker protection"""

        if self.state == CircuitState.OPEN:
            # Check if timeout elapsed
            if datetime.now() - self.last_failure_time > timedelta(seconds=self.timeout):
                self.state = CircuitState.HALF_OPEN
                print("? Circuit breaker: HALF_OPEN (testing recovery)")
            else:
                raise RuntimeError("Circuit breaker OPEN - service unavailable")

        try:
            result = func(*args, **kwargs)

            # Success - reset if recovering
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
                print("? Circuit breaker: CLOSED (service recovered)")

            return result

        except self.expected_exception as e:
            self.failure_count += 1
            self.last_failure_time = datetime.now()

            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
                print(f"? Circuit breaker: OPEN (threshold {self.failure_threshold} reached)")

            raise

# Usage
circuit_breaker = CircuitBreaker(failure_threshold=5, timeout=60)

def generate_protected(prompt: str):
    """Generate with circuit breaker protection"""
    return circuit_breaker.call(llm.generate, prompt)

# If llm.generate fails 5 times, circuit breaker opens
# Requests fail fast for 60 seconds
# Then one test request (half-open)
# If successful, normal operation resumes

import time

class RateLimiter:
    """
    Token bucket rate limiter

    Limits requests per second to prevent overload
    """

    def __init__(self, max_requests: int, time_window: float = 1.0):
        self.max_requests = max_requests
        self.time_window = time_window
        self.tokens = max_requests
        self.last_update = time.time()

    def acquire(self, tokens: int = 1) -> bool:
        """Try to acquire tokens, return True if allowed"""

        now = time.time()
        elapsed = now - self.last_update

        # Refill tokens based on elapsed time
        self.tokens = min(
            self.max_requests,
            self.tokens + (elapsed / self.time_window) * self.max_requests
        )
        self.last_update = now

        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        else:
            return False

    def wait_for_token(self, tokens: int = 1):
        """Wait until token is available"""
        while not self.acquire(tokens):
            time.sleep(0.1)

# Usage
rate_limiter = RateLimiter(max_requests=100, time_window=1.0)

@app.post("/generate")
async def generate(request: GenerateRequest):
    # Check rate limit
    if not rate_limiter.acquire():
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded (100 req/sec)"
        )

    # Process request
    result = llm.generate(request.prompt)
    return result

def generate_with_fallback(prompt: str) -> str:
    """
    Try multiple strategies before failing

    Strategy 1: Primary model (Llama-3-8B)
    Strategy 2: Cached response (if available)
    Strategy 3: Simpler model (Phi-2)
    Strategy 4: Template response
    """

    # Try primary model
    try:
        return llm_primary.generate(prompt)

    except Exception as e:
        print(f"??  Primary model failed: {e}")

        # Fallback 1: Check cache
        cached_response = cache.get(prompt)
        if cached_response:
            print("? Returning cached response")
            return cached_response

        # Fallback 2: Try simpler model
        try:
            print("? Falling back to Phi-2")
            return llm_simple.generate(prompt)

        except Exception as e2:
            print(f"??  Fallback model also failed: {e2}")

            # Fallback 3: Template response
            return (
                "I apologize, but I'm unable to process your request right now. "
                "Please try again in a few minutes, or contact support if the issue persists."
            )

# User never sees "Internal Server Error"
# They always get SOME response

import signal

class TimeoutError(Exception):
    pass

def timeout_handler(signum, frame):
    raise TimeoutError("Request timed out")

def generate_with_timeout(prompt: str, timeout_seconds: int = 30):
    """Generate with timeout"""

    # Set timeout
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout_seconds)

    try:
        result = llm.generate(prompt)

        # Cancel timeout
        signal.alarm(0)
        return result

    except TimeoutError:
        print(f"? Request timed out after {timeout_seconds}s")
        return "Request timed out. Please try a shorter prompt."

# Or using asyncio
import asyncio

async def generate_with_timeout_async(prompt: str, timeout_seconds: int = 30):
    """Generate with async timeout"""

    try:
        result = await asyncio.wait_for(
            llm.generate_async(prompt),
            timeout=timeout_seconds
        )
        return result

    except asyncio.TimeoutError:
        return "Request timed out. Please try a shorter prompt."

from fastapi import FastAPI, HTTPException
from circuitbreaker import CircuitBreaker, CircuitBreakerError

app = FastAPI()

# Initialize components
circuit_breaker = CircuitBreaker(failure_threshold=5, timeout=60)
rate_limiter = RateLimiter(max_requests=100, time_window=1.0)
cache = ResponseCache(ttl=3600)

@app.post("/generate")
@retry_with_backoff(max_retries=3)
async def generate(request: GenerateRequest):
    """
    Generate with full error handling:
    - Rate limiting
    - Circuit breaker
    - Retry with backoff
    - Timeout
    - Fallback strategies
    - Caching
    """

    # Rate limiting
    if not rate_limiter.acquire():
        raise HTTPException(status_code=429, detail="Rate limit exceeded")

    # Check cache first
    cached = cache.get(request.prompt)
    if cached:
        return {"text": cached, "cached": True}

    try:
        # Circuit breaker protection
        result = circuit_breaker.call(
            generate_with_timeout,
            request.prompt,
            timeout_seconds=30
        )

        # Cache successful response
        cache.set(request.prompt, result)

        return {"text": result, "status": "success"}

    except CircuitBreakerError:
        # Circuit breaker open - return fallback
        return {
            "text": "Service temporarily unavailable. Using cached response.",
            "status": "degraded",
            "fallback": True
        }

    except TimeoutError:
        raise HTTPException(status_code=504, detail="Request timed out")

    except Exception as e:
        # Log error
        logger.error(f"Generation failed: {e}")

        # Return graceful error
        return {
            "text": "I apologize, but I'm unable to process your request.",
            "status": "error",
            "fallback": True
        }

Request 1 (uncached): $0.00002038
Requests 2-11 (cached): $0.00000414 average

Total: $0.00010031 for 11 requests
Average: $0.0000091 per request

GCP L4 GPU: $0.45/hour = $328/month

Total monthly cost:
- Infrastructure: $328
- API costs: $2.73
- Total: $330.73/month for 10,000 requests/day

OpenAI GPT-4:
- Input: $0.03 per 1K tokens
- Output: $0.06 per 1K tokens
- Average request: 100 tokens in + 100 tokens out = $0.009
- 10,000 requests/day = $90/day = $2,700/month

Savings: $2,369/month (88% cheaper!)

Cost per request: $0.009
Daily: $450
Monthly: $13,500
Annual: $162,000

Infrastructure: $328/month
API costs (with optimizations): $13.65/month
Monthly total: $341.65
Annual: $4,100

Savings: $157,900/year (97%)

Setup time: 2 days engineering ($2,000)
Maintenance: 4 hours/month ($200/month)

Year 1:
  Savings: $157,900
  Costs: $2,000 setup + $2,400 maintenance = $4,400
  Net: $153,500 saved

ROI: 3,500% in year 1

OpenAI GPT-4: $1,350,000/year
vLLM solution: $41,000/year

Savings: $1,309,000/year (97%)

# src/services/startup_service.py
async def preload_daily_data():
    """Background job that generates brief and caches in database."""

    # 1. Fetch all data in parallel (LangGraph orchestration)
    data = await langgraph_orchestrator.fetch_all_sources()

    # 2. Generate AI brief (ReAct agent with RAG)
    brief = await brief_generator.generate(
        emails=data['emails'],
        calendar=data['calendar'],
        news=data['news'],
        weather=data['weather']
    )

    # 3. Cache everything in SQLite
    await db.set_cache('daily_brief_data', brief.to_dict(), ttl=3600)  # 1 hour TTL
    await db.set_cache('news_data', data['news'], ttl=3600)
    await db.set_cache('emails_data', data['emails'], ttl=3600)

    logger.info("? All data preloaded and cached")

# src/ui/components/daily_brief.py
def render_daily_brief_section():
    """UI just reads from cache - no AI processing!"""

    # Fast read from database (milliseconds, not seconds)
    if 'data' in st.session_state and st.session_state.data.get('daily_brief'):
        brief_data = st.session_state.data['daily_brief']
        _display_persisted_brief(brief_data)  # Instant!
    else:
        st.info("Run `make preload` to generate your first brief.")

# On macOS (what I use)
brew install ollama

# Start the service
ollama serve

# Pull the models I chose
ollama pull qwen2.5:7b         # Main LLM - fast on my M3 Mac
ollama pull nomic-embed-text   # For RAG embeddings

# Quick test
PYTHONPATH=. python -c "
import asyncio
from src.services.ollama_service import get_ollama_service

async def test():
    ollama = get_ollama_service()
    result = await ollama.generate('Explain RAG in one sentence.')
    print(result)

asyncio.run(test())
"

# Output I got:
# RAG (Retrieval-Augmented Generation) enhances LLM responses by retrieving
# relevant information from a knowledge base before generating answers.

# ? My first attempt (brittle)
async def get_daily_data():
    news = await fetch_hackernews()
    weather = await fetch_weather()
    # Later I wanted to add RSS feeds... had to modify this function
    # Then I wanted Slack... modified again
    # This was getting messy fast!

# src/connectors/hackernews.py
class HackerNewsConnector:
    """Fetches top stories from HackerNews API."""

    async def execute_async(self, max_stories: int = 10):
        """The main method MCP will call."""
        # 1. Fetch top story IDs
        response = await self.client.get(
            "https://hacker-news.firebaseio.com/v0/topstories.json"
        )
        story_ids = response.json()[:max_stories]

        # 2. Fetch each story (I fetch these in parallel for speed)
        articles = []
        for story_id in story_ids:
            story = await self._fetch_story(story_id)
            articles.append(self._convert_to_article(story))

        return articles

# src/services/mcp_server.py
class MCPServer:
    """The tool registry that AI agents query."""

    def _register_tools(self):
        # Register HackerNews
        self.tools["fetch_hackernews"] = MCPTool(
            name="fetch_hackernews",
            description="Fetch top tech stories from HackerNews with scores and comments",
            parameters={
                "max_stories": {
                    "type": "integer",
                    "description": "How many stories to fetch (1-30)",
                    "default": 10
                }
            },
            executor=HackerNewsConnector()
        )

# I tested if the AI could discover my tools
PYTHONPATH=. python -c "
from src.services.mcp_server import get_mcp_server

mcp = get_mcp_server()
print('Available tools:')
for tool in mcp.list_tools():
    print(f'  ? {tool[\"name\"]}: {tool[\"description\"]}')
"

# Output I got:
# Available tools:
#   ? fetch_hackernews: Fetch top tech stories from HackerNews...
#   ? get_current_weather: Get current weather conditions...
#   ? fetch_rss_feeds: Fetch articles from configured RSS feeds...

# src/services/rag_service.py
class RAGService:
    """Semantic memory using ChromaDB."""

    def __init__(self):
        # Initialize ChromaDB (stores on disk)
        self.client = chromadb.Client(Settings(
            persist_directory="./data/chroma_data"
        ))

        # Create collection for my articles
        self.collection = self.client.get_or_create_collection(
            name="daily_minutes"
        )

        # Ollama for creating embeddings
        self.ollama = get_ollama_service()

    async def add_document(self, content: str, metadata: dict):
        """Store a document with its vector embedding."""

        # 1. Convert text to vector (this is the magic!)
        embedding = await self.ollama.create_embeddings(content)

        # 2. Store in ChromaDB with metadata
        self.collection.add(
            documents=[content],
            embeddings=[embedding],
            metadatas=[metadata],
            ids=[hashlib.md5(content.encode()).hexdigest()]
        )

    async def search(self, query: str, max_results: int = 5):
        """Semantic search - find by meaning!"""

        # 1. Convert query to vector
        query_embedding = await self.ollama.create_embeddings(query)

        # 2. Find similar documents (cosine similarity)
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=max_results
        )

        return results

# I stored an article about EU AI regulations
await rag.add_document(
    content="European Union announces comprehensive AI safety regulations "
            "focusing on transparency, accountability, and privacy protection.",
    metadata={"type": "article", "topic": "ai_safety"}
)

# Later, I searched using different words
results = await rag.search("privacy rules for artificial intelligence")

# 1. News articles (for historical context)
await rag.add_article(article)

# 2. Action items from emails
await rag.add_todo(
    "Complete security training by Nov 15",
    source="email",
    priority="high"
)

# 3. Meeting context
await rag.add_document(
    "Q4 Planning Meeting - need to prepare budget estimates",
    metadata={"type": "meeting", "date": "2025-02-01"}
)

# 4. User preferences (this feeds into RLHF later!)
await rag.add_document(
    "User marked 'AI safety' topics as important",
    metadata={"type": "preference", "category": "ai_safety"}
)

# ? First attempt - no thinking!
async def generate_brief():
    news = await fetch_all_news()  # Fetches everything
    summary = await llm.generate(f"Summarize: {news}")
    return summary

# src/agents/react_agent.py
class ReActAgent:
    """Agent that thinks before acting."""

    async def run(self, goal: str):
        """Execute goal using ReAct loop."""
        steps = []
        observations = []

        for step_num in range(1, self.max_steps + 1):
            # 1. THOUGHT: Ask AI what to do next
            thought = await self._generate_thought(goal, steps, observations)

            # Check if we're done
            if "FINAL ANSWER" in thought:
                return self._extract_answer(thought)

            # 2. ACTION: Parse what action to take
            action = self._parse_action(thought)
            # Example: {"action": "call_tool", "tool": "fetch_hackernews"}

            # 3. EXECUTE: Run the action via MCP
            observation = await self._execute_action(action)
            observations.append(observation)

            # Record this step for debugging
            steps.append({
                "thought": thought,
                "action": action,
                "observation": observation
            })

        return {"steps": steps, "answer": "Max steps reached"}

async def _generate_thought(self, goal, steps, observations):
    """Generate next reasoning step."""

    prompt = f"""Goal: {goal}

Previous steps:
{self._format_steps(steps)}

Available actions:
- query_rag(query): Search my semantic memory
- call_tool(name, params): Execute an MCP tool
- FINAL ANSWER: When you have everything needed

Think step-by-step. What should I do next?

Format your response as:
THOUGHT: <your reasoning>
ACTION: <action to take>
"""

    return await self.ollama.generate(prompt, temperature=0.7)

? Goal: Generate my daily brief

Step 1:
  ? THOUGHT: I need to gather news, check weather, and see user preferences
  ? ACTION: call_tool("fetch_hackernews", max_stories=10)
  ?? OBSERVATION: Fetched 10 articles about AI, privacy, and tech

Step 2:
  ? THOUGHT: Got news. User preferences would help prioritize.
  ? ACTION: query_rag("user interests and preferences")
  ?? OBSERVATION: User cares about AI safety, security, privacy

Step 3:
  ? THOUGHT: Should filter articles to user's interests
  ? ACTION: call_tool("get_current_weather", location="Seattle")
  ?? OBSERVATION: 70°F, Partly cloudy

Step 4:
  ? THOUGHT: I have news (filtered by user interests), weather. Ready to generate.
  ? ACTION: FINAL ANSWER
  ? Generated personalized brief highlighting AI safety articles

? "Newsletter: 10 CSS Tips" ? Importance: 0.5
? "URGENT: Production outage!" ? Importance: 0.5

# src/models/email.py
class ImportanceScoringMixin:
    """Learn from user feedback."""

    importance_score: float = 0.5  # AI's base score
    boost_labels: Set[str] = set()  # Words user marked important
    filter_labels: Set[str] = set()  # Words user wants to skip

    def apply_rlhf_boost(self, content_text: str) -> float:
        """Adjust score based on learned preferences."""
        adjusted = self.importance_score
        content_lower = content_text.lower()

        # Boost if content matches important keywords
        for label in self.boost_labels:
            if label.lower() in content_lower:
                adjusted += 0.1  # Bump up priority!

        # Penalize if content matches skip keywords
        for label in self.filter_labels:
            if label.lower() in content_lower:
                adjusted -= 0.2  # Push down priority!

        # Keep in valid range [0, 1]
        return max(0.0, min(1.0, adjusted))

# User sees an email
for email in emails:
    col1, col2, col3 = st.columns([8, 1, 1])

    with col1:
        st.write(f"**{email.subject}**")
        st.info(email.snippet)

    with col2:
        if st.button("?", key=f"important_{email.id}"):
            # Extract what made this important
            keywords = await extract_keywords(email.subject + email.body)
            # Add to boost labels
            user_profile.boost_labels.update(keywords)
            st.success(f"? Learned: You care about {', '.join(keywords)}")

    with col3:
        if st.button("?", key=f"skip_{email.id}"):
            # Learn to deprioritize these
            keywords = await extract_keywords(email.subject)
            user_profile.filter_labels.update(keywords)
            st.success(f"? Will deprioritize: {', '.join(keywords)}")

# ? Sequential execution - SLOW!
news = await fetch_news()      # 5 seconds
emails = await fetch_emails()  # 3 seconds
calendar = await fetch_calendar()  # 2 seconds
weather = await fetch_weather()  # 1 second
# Total: 11 seconds just waiting! ?

# src/services/langgraph_orchestrator.py
from langgraph.graph import StateGraph, END

class LangGraphOrchestrator:
    def _create_workflow(self):
        """Define my workflow graph."""
        workflow = StateGraph(WorkflowState)

        # Add nodes (processing steps)
        workflow.add_node("analyze", self._analyze_request)
        workflow.add_node("fetch_news", self._fetch_news)
        workflow.add_node("fetch_emails", self._fetch_emails)
        workflow.add_node("fetch_calendar", self._fetch_calendar)
        workflow.add_node("search_rag", self._search_context)
        workflow.add_node("generate_summary", self._generate_summary)

        # Define edges (execution flow)
        workflow.set_entry_point("analyze")

        # Parallel fetch (all happen at once!)
        workflow.add_edge("analyze", "fetch_news")
        workflow.add_edge("analyze", "fetch_emails")
        workflow.add_edge("analyze", "fetch_calendar")

        # All converge to RAG search
        workflow.add_edge("fetch_news", "search_rag")
        workflow.add_edge("fetch_emails", "search_rag")
        workflow.add_edge("fetch_calendar", "search_rag")

        # Sequential processing
        workflow.add_edge("search_rag", "generate_summary")
        workflow.add_edge("generate_summary", END)

        return workflow.compile()

async def _fetch_news(self, state: WorkflowState):
    """Fetch news in parallel."""
    try:
        articles = await self.mcp.execute_tool(
            "fetch_hackernews",
            {"max_stories": 10}
        )
        state["news_articles"] = articles
    except Exception as e:
        state["errors"].append(f"News fetch failed: {e}")
        state["news_articles"] = []

    return state

async def _search_context(self, state: WorkflowState):
    """Search RAG for relevant context."""
    query = state["user_request"]
    results = await self.rag.search(query, max_results=5)

    # Build context string
    context = "\n".join([r['content'] for r in results])
    state["context"] = context

    return state

# Execute the complete workflow
result = await orchestrator.run("Generate my daily brief")

# I get back:
{
    "news_articles": [...],      # 10 articles
    "emails": [...],              # 5 unread
    "calendar_events": [...],     # 3 events today
    "context": "...",             # RAG context
    "summary": "...",             # Generated brief
    "processing_time": 5.2        # Seconds (not 11!)
}

# src/services/llm_factory.py
def get_llm_service():
    """Factory pattern - works with any LLM provider."""
    provider = os.getenv("LLM_PROVIDER", "ollama")

    if provider == "ollama":
        return OllamaService(
            base_url=os.getenv("OLLAMA_BASE_URL", "http://localhost:11434"),
            model=os.getenv("OLLAMA_MODEL", "qwen2.5:7b")
        )
    elif provider == "openai":
        return OpenAIService(
            api_key=os.getenv("OPENAI_API_KEY"),
            model=os.getenv("OPENAI_MODEL", "gpt-4-turbo")
        )
    elif provider == "google":
        # Like in my previous Vertex AI article!
        return VertexAIService(
            project_id=os.getenv("GCP_PROJECT_ID"),
            model="gemini-1.5-flash"
        )

    raise ValueError(f"Unknown provider: {provider}")

# All services implement the same interface:
class BaseLLMService:
    async def generate(self, prompt: str, **kwargs) -> str:
        """Generate text from prompt."""
        raise NotImplementedError

    async def create_embeddings(self, text: str) -> List[float]:
        """Create vector embeddings."""
        raise NotImplementedError

? "Today's news features a mix of technology updates and various topics."

# ? Better prompt with strict rules
prompt = f"""Generate a daily brief following these STRICT rules:

PRIORITY ORDER (most important first):
1. Urgent emails or action items
2. Today's calendar events
3. Market/business news
4. Tech news

TLDR FORMAT (exactly 3 bullets, be SPECIFIC):
* Bullet 1: Most urgent email/action (include WHO, WHAT, WHEN)
   Example: "Client escalation from Acme Corp affecting 50K users - response needed by 2pm"

* Bullet 2: Most important calendar event today (include TIME and WHAT TO PREPARE)
   Example: "2pm: Board meeting - prepare Q4 revenue slides"

* Bullet 3: Top market/business news (include NUMBERS/SPECIFICS)
   Example: "Federal Reserve raises rates 0.5% to 5.25% - affects tech hiring"

AVOID THESE PHRASES (they're too vague):
? "mix of updates"
? "various topics"
? "continues to make progress"
? "interesting developments"

USE SPECIFIC DETAILS:
? Names (people, companies)
? Numbers (percentages, dollar amounts, deadlines)
? Times (when something happened or needs to happen)

Content to summarize:
{content}

Generate: TLDR (3 bullets), Summary (5-6 detailed sentences), Key Insights (5 bullets)
"""

? • Critical email... • Meeting at 2pm... • Market news...

# ? Doesn't work
st.info(tldr)

# ? Works!
tldr_lines = [line.strip() for line in tldr.split('\n') if line.strip()]
for bullet in tldr_lines:
    st.markdown(bullet)

? TLDR bullet 1: "OpenAI releases GPT-5" (who cares?)
   TLDR bullet 2: "Crypto market surges" (not relevant to me)
   TLDR bullet 3: "Client escalation requires response" (BURIED!)

# src/services/brief_scheduler.py
async def _generate_daily_brief(emails, calendar, news, weather):
    """Generate prioritized daily brief with structured prompt."""

    # Separate market vs tech news (market is higher priority)
    market_news = [n for n in news if 'market' in n.tags]
    tech_news = [n for n in news if 'market' not in n.tags]

    # Sort emails by RLHF-boosted importance score
    important_emails = sorted(
        emails,
        key=lambda e: e.apply_rlhf_boost(e.subject + e.snippet),
        reverse=True
    )[:5]  # Top 5 only

    # Build structured prompt with clear priority
    prompt = f"""
**SECTION 1: IMPORTANT EMAILS (HIGHEST PRIORITY - use for TLDR bullet #1)**
{format_emails(important_emails)}

**SECTION 2: TODAY'S CALENDAR (SECOND PRIORITY - use for TLDR bullet #2)**
{format_calendar(calendar)}

**SECTION 3: MARKET NEWS (THIRD PRIORITY - use for TLDR bullet #3)**
{format_market_news(market_news)}

**SECTION 4: TECH NEWS (LOWEST PRIORITY - summarize briefly)**
{format_tech_news(tech_news)}

**SECTION 5: WEATHER**
{format_weather(weather)}

Generate a daily brief following this EXACT priority order:
1. Email action items FIRST
2. Calendar events SECOND
3. Market/business news THIRD
4. Tech news LAST (brief mention only)

TLDR must have EXACTLY 3 bullets using content from sections 1, 2, 3 (not section 4).
"""

    return await llm.generate(prompt)

Query: "AI safety regulations"
Result: Article about "safe AI models for healthcare" (wrong context!)

# Store with rich metadata
await rag.add_document(
    content=article.title,
    metadata={
        "type": "article",
        "category": "ai_safety",  # For filtering!
        "tags": ["regulation", "eu", "policy"],
        "date": "2025-01-28",
        "importance": "high"
    }
)

# Search with filters
results = await rag.search(
    "AI regulations",
    filter_metadata={
        "category": "ai_safety",
        "importance": "high"
    }
)

  # Current: Simple keyword match
  if "security" in email:
      score += 0.1

  # Better: Contextual understanding
  if embeddings_similar(email, user.important_emails):
      score += contextual_boost  # Uses semantic similarity!

# src/connectors/slack.py
class SlackConnector:
    """Fetch recent messages from Slack channels."""

    async def execute_async(self, channel: str, max_messages: int = 10):
        # 1. Connect to Slack API
        client = WebClient(token=os.getenv("SLACK_BOT_TOKEN"))

        # 2. Fetch recent messages
        response = await client.conversations_history(
            channel=channel,
            limit=max_messages
        )

        # 3. Convert to standard format
        messages = []
        for msg in response['messages']:
            messages.append(SlackMessage(
                text=msg['text'],
                user=msg['user'],
                channel=channel,
                timestamp=msg['ts']
            ))

        return messages

# src/services/mcp_server.py
def _register_tools(self):
    # ... existing tools ...

    # Add Slack
    self.tools["get_slack_messages"] = MCPTool(
        name="get_slack_messages",
        description="Fetch recent Slack messages from a channel",
        parameters={
            "channel": {"type": "string", "description": "Channel name"},
            "max_messages": {"type": "integer", "default": 10}
        },
        executor=SlackConnector()
    )

# Your ReAct agent will now see:
# "Available tools: fetch_hackernews, get_slack_messages, ..."
# No changes needed to ReAct logic!

# src/services/brief_scheduler.py
prompt = f"""
**IMPORTANT EMAILS**: {emails}
**CALENDAR**: {calendar}
**SLACK HIGHLIGHTS**: {slack_messages}  # New!
**NEWS**: {news}

Generate brief prioritizing: Email > Calendar > Slack > News
"""

# .env.development
LLM_PROVIDER=ollama
LLM_OLLAMA_BASE_URL=http://localhost:11434
LLM_OLLAMA_MODEL=qwen2.5:7b
DATABASE_URL=sqlite:///./data/daily_minutes.db
REDIS_URL=redis://localhost:6379

# .env.production
LLM_PROVIDER=openai  # Just change this line!
OPENAI_API_KEY=sk-...
OPENAI_MODEL=gpt-4-turbo
DATABASE_URL=postgresql://...  # Scalable DB
REDIS_URL=redis://prod-cluster:6379  # Distributed cache

# src/services/llm_factory.py
def get_llm_service():
    """Factory pattern - works with any LLM."""
    provider = os.getenv("LLM_PROVIDER", "ollama")

    if provider == "ollama":
        return OllamaService()
    elif provider == "openai":
        return OpenAIService()
    elif provider == "anthropic":
        return ClaudeService()
    elif provider == "google":
        return VertexAIService()  # Like in my previous article!

    raise ValueError(f"Unknown provider: {provider}")

# Test MCP tool registration
pytest tests/unit/test_mcp_server.py -v

# Test RAG semantic search
pytest tests/unit/test_rag_service.py -v

# Test ReAct reasoning
pytest tests/unit/test_react_agent.py -v

# Test RLHF scoring
pytest tests/unit/test_rlhf_scoring.py -v

# Run all unit tests
pytest tests/unit/ -v
# 516 passed in 45.23s ?

# tests/integration/test_brief_quality.py
async def test_tldr_has_three_bullets():
    """TLDR must have exactly 3 bullets."""
    brief = await db.get_cache('daily_brief_data')
    tldr = brief.get('tldr', '')

    bullets = [line for line in tldr.split('\n') if line.strip().startswith('•')]

    assert len(bullets) == 3, f"Expected 3 bullets, got {len(bullets)}"
    assert "email" in bullets[0].lower() or "urgent" in bullets[0].lower()
    assert "calendar" in bullets[1].lower() or "meeting" in bullets[1].lower()

async def test_no_generic_phrases():
    """Brief should not contain vague phrases."""
    brief = await db.get_cache('daily_brief_data')
    summary = brief.get('summary', '')

    bad_phrases = ["mix of updates", "various topics", "continues to"]
    for phrase in bad_phrases:
        assert phrase not in summary.lower(), f"Found generic phrase: {phrase}"

# 1. Fetch data and generate brief
make preload

# Output I see:
# ? Fetching news from HackerNews... (10 articles)
# ? Fetching weather... (70°F, Sunny)
# ? Analyzing articles with AI... (15 articles)
# ? Generating daily brief... (Done in 18.3s)
# ? Brief saved to database

# 2. Launch UI
streamlit run src/ui/streamlit_app.py

# 3. Check brief quality
# - Is TLDR specific? (not vague)
# - Are priorities correct? (email > calendar > news)
# - Are action items extracted? (from emails)
# - Did RLHF work? (boosted my preferences)

# Clone my repo
git clone https://github.com/bhatti/daily-minutes
cd daily-minutes

# Install dependencies
pip install -r requirements.txt

# Setup Ollama
ollama pull qwen2.5:7b
ollama pull nomic-embed-text

# Generate your first brief
make preload

# Launch dashboard
streamlit run src/ui/streamlit_app.py