// Copyright 2024-2026 Lux Industries Inc. All rights reserved. // Use of this source code is governed by the Apache 2.0 // license that can be found in the LICENSE file. package api // The one assembly of this engine over one store. // // Everything a transaction touches is built here, in one function, and cmd/amld // and every test in this package call it. That is not tidiness — it is the fix // for the defect that produced the last review. // // A monitoring engine is a dozen planes joined to one ingest path. When the // deployment joins them in one place and the tests join them in another, the two // drift, and the drift is invisible: the suite is green because it is proving // properties of the arrangement it built. The record fingerprint was written to a // struct field on a shelf the tests wired and never to a column on the shelf // cmd/amld wires, so every retry conflicted permanently in production while every // test passed. The bug was not in the ledger. It was that there were two wirings. // // So there is one, and a test reads cmd/amld to keep it that way. A plane added // here is wired for the deployment and for the suite at the same moment; a plane // not added here does not exist for either. import ( "context" "fmt" "time" "github.com/hanzoai/base/core" "github.com/luxfi/aml/pkg/anomaly" "github.com/luxfi/aml/pkg/cases" "github.com/luxfi/aml/pkg/dictionary" "github.com/luxfi/aml/pkg/engine" "github.com/luxfi/aml/pkg/history" "github.com/luxfi/aml/pkg/lists" "github.com/luxfi/aml/pkg/models" "github.com/luxfi/aml/pkg/receipt" "github.com/luxfi/aml/pkg/reference" "github.com/luxfi/aml/pkg/retention" "github.com/luxfi/aml/pkg/screen" "github.com/luxfi/aml/pkg/suppress" "github.com/luxfi/aml/pkg/token" "github.com/luxfi/aml/pkg/topology" "github.com/luxfi/aml/pkg/types" "github.com/luxfi/aml/pkg/velocity" "github.com/luxfi/aml/pkg/watch" ) // Deployment is what differs between two installations of this engine. // // Everything NOT here — which store each plane sits on, which planes exist at // all, how they are joined, what runs on a cadence — is the same everywhere and // is [Wire]'s. A field belongs here only if a deployment genuinely answers it // differently, and each one below does: who a caller is, which IAM application // this is, what the institution detects, the key material it holds, the // designations it screens against, its business day, and its rates. type Deployment struct { // Identity resolves which tenant a request is authenticated to act on, and // which subject is deciding. cmd/amld supplies IAMIdentity; a test supplies a // fixed caller. Identity Identity // ClientID is this deployment's IAM application: the audience every token is // pinned to, and what the console reads its own identity from. ClientID string // Rules is the detection library. Installing it is what checks that every // rule's evidence exists, so it happens after the providers are attached and // a failure refuses the whole assembly. Rules []types.Rule // Keys is the tokenisation keyring over the KMS-held root. Keys *token.Keyring // Screen is the designation store, and Readiness is which publisher is behind // it. They are the deployment's because something outside this assembly fills // them on a cadence. Screen *screen.Store Readiness *screen.Readiness // Zone is the business day calendar rules are answered in. Nil means UTC. Zone *time.Location // Rate converts an amount to the unit thresholds are stated in. Rate reference.Rates // Jurisdictions is the sanctioned-jurisdiction reference the rules read. Jurisdictions reference.Jurisdictions // Live lets the behavioural model contribute to a verdict. // // It is stated from the LIVE side so that the zero value is shadow. A // detection has to be testable before it is activated: a new deployment // scores, learns and publishes at GET /v1/aml/anomaly what it WOULD have // alerted on, and contributes nothing to any transaction's outcome until // somebody has read that. Written as `Shadow bool` the same field would arm // the model for anyone who did not think about it, which is the one direction // a default must never fail in. Live bool // Limit is the reporting limit transactions are judged against. Zero takes // the fallback. Limit float64 } // Wire assembles the whole engine over one store. // // The order is the design. The collections exist before any shelf opens over // them; the shelves exist before the joins; the engine is built over the evidence // providers; the rule library is installed LAST, because installing a rule is // what checks that its evidence exists and that check is worth nothing if it runs // before the providers are attached. // // It returns a Handler ready to Register, and it has already put on the app // everything that has to run on a cadence for the durable state to stay honest. func Wire(app core.App, d Deployment) (*Handler, error) { // Every collection this engine writes to, created idempotently, in one list. // A shelf whose column does not exist does not fail: it reads back a zero // value, which is a plausible answer, which is how a record fingerprint came // to be a struct field nothing stored. for _, ensure := range []func(core.App) error{ history.Ensure, retention.Ensure, receipt.Ensure, cases.Ensure, EnsureAlerts, lists.Ensure, suppress.Ensure, watch.Ensure, dictionary.Ensure, models.Ensure, } { if err := ensure(app); err != nil { return nil, fmt.Errorf("a record plane cannot be created: %w", err) } } events := history.NewBase(app) deny := lists.NewBase(app) silence := suppress.NewBase(app) monitor := watch.NewBase(app) monitor.Cover = silence catalog := dictionary.NewBase(app) // The behavioural plane. Sliding aggregates are the substrate every // behavioural measure reads; the model reads them to score whether a // transaction is unusual for the entity that made it. windows := velocity.New(velocity.Config{}) model, err := anomaly.New(anomaly.Config{Shadow: !d.Live}, windows) if err != nil { return nil, fmt.Errorf("the behavioural plane cannot be built: %w", err) } // Half the machine, at most, for every study and every replay together — the // other half is ingest's, and a transaction that cannot be recorded cannot be // processed. ONE budget for the process, because the CPU is one machine // however many planes ask for it. See topology.Budget. cores := topology.NewBudget(0) study := models.NewBase(app) study.Model = model study.Cores = cores // A model planted after a rollout or an admission asks the model plane what // this tenant last adopted, so an adopted control cannot go quiet without // somebody deciding it should. model.SetAdopted(study.Adopted) eng := engine.New(engine.Providers{ History: events, Lists: deny, Screen: d.Screen, Reference: d.Jurisdictions, Rate: d.Rate, Zone: d.Zone, }) eng.SetScorer(model) if err := eng.SetRules(d.Rules); err != nil { return nil, fmt.Errorf("the detection library cannot be installed: %w", err) } h := &Handler{ Identity: d.Identity, ClientID: d.ClientID, Engine: eng, Cases: cases.NewBase(app), Alerts: NewAlertStoreBase(app), Screen: d.Screen, Readiness: d.Readiness, History: events, Rate: d.Rate, Records: retention.NewBase(app), Receipts: receipt.NewBase(app), Keys: d.Keys, Velocity: windows, Anomaly: model, Cores: cores, Limit: d.limit(), Planes: Planes{ Lists: deny, Suppress: silence, Watch: monitor, Dictionary: catalog, Models: study, }, } // The model plane reads a tenant's history through the same join a rule replay // uses, so a study of the model and a study of a rule see the same events. study.History = Replayed{H: h} tend(app, h, catalog) return h, nil } // limit is the reporting limit this deployment states, or the fallback. func (d Deployment) limit() float64 { if d.Limit > 0 { return d.Limit } return reportLimit } // tend puts on the app everything that has to run on a cadence for the durable // state to stay honest. // // It is part of the assembly and not the caller's, for the same reason the // collections are: a flush nobody runs is a statistic that only ever reads as // pending, and a disposal nobody runs is a ledger that grows forever. Both are // properties of what was wired, so both are wired here. func tend(app core.App, h *Handler, catalog *dictionary.Shelf) { // The field catalog accumulates in memory and is written on a cadence and at // shutdown. flush := func() { if err := catalog.Flush(context.Background()); err != nil { app.Logger().Error("field catalog flush failed", "error", err) } } app.Cron().Add("dictionary-flush", "*/5 * * * *", flush) app.OnTerminate().BindFunc(func(te *core.TerminateEvent) error { flush() return te.Next() }) // Destroy records whose retention period has run out, daily. retention.Cron(app, h.Records) // And drop the receipts that can no longer affect an aggregate. It runs at a // different hour from the retention disposal so two bulk deletes do not meet // on one SQLite writer. app.Cron().Add("receipt-dispose", "0 4 * * *", func() { n, err := h.Receipts.Dispose(context.Background(), time.Now().UTC()) if err != nil { app.Logger().Error("receipt disposal failed", "error", err) return } app.Logger().Info("receipts disposed", "count", n) }) }