"""PRAJA -- the citizen's agent (one per household).
Works only for the family. Finds every scheme they qualify for, prepares the
applications, gets each adult's own approval, files with each RAJA, closes
evidence gaps by asking other agents first, picks options within the family's
constraints, follows each claim until money lands, and complains when it
doesn't. It never submits without approval and never asserts facts.
The language layer (BHASHINI / an LLM) only turns these decisions into
speech and SMS. It decides nothing; see voice.py.
"""
from __future__ import annotations
import itertools
from .protocol import Envelope
from .rules import evaluate
from .voice import say
_cp = itertools.count(41207)
class Praja:
def __init__(self, household_id, members, head, schemes, registries, ledger, mandate, constraints=None):
self.id = f"PRAJA:{household_id}"
self.members, self.head = members, head
self.schemes, self.reg, self.ledger, self.mandate = schemes, registries, ledger, mandate
self.constraints = constraints or {}
self.outbox: list[str] = [] # what the family hears / reads
self.claims: dict[str, dict] = {} # ref -> {scheme_id, person, status}
self.peers: dict[str, str] = {} # role -> agent id (college, bank)
def tell(self, key, **kw):
self.outbox.append(say(key, **kw))
def _ask(self, person, predicate, arg):
return self.reg.ask(self.id, self.mandate.mandate_id, person, predicate, arg)
# --- 1. Find ------------------------------------------------------
def discover(self):
found = [evaluate(s, p, self._ask) for s in self.schemes.values() for p in self.members]
# Contributory schemes (e.g. Atal Pension) are explained, never auto-applied:
# joining one costs the family money each month, so it is the family's call.
self.inform_only = [r for r in found if r.status == "eligible" and self.schemes[r.scheme_id].get("inform_only")]
found = [r for r in found if r not in self.inform_only]
eligible = [r for r in found if r.status == "eligible"]
possibly = [r for r in found if r.status == "possibly"]
# Mutually exclusive benefits (e.g. two state pensions): keep the better one.
best: dict[tuple, object] = {}
for r in eligible:
g = self.schemes[r.scheme_id].get("exclusive_group")
if not g:
continue
k = (r.person, g)
if k not in best or self._value(r) > self._value(best[k]):
best[k] = r
chosen = [r for r in eligible if not self.schemes[r.scheme_id].get("exclusive_group")
or best[(r.person, self.schemes[r.scheme_id]["exclusive_group"])] is r]
set_aside = [r for r in eligible if r not in chosen]
# Near misses for the head of household: failed exactly one rule -> tell her why.
near = [r for r in found if r.person == self.head and r.status == "not_eligible"
and sum(1 for t in r.trace if t[1] == "not_met") == 1
and not any(t[1] == "skipped" or t[2] == "no record found" for t in r.trace)]
return chosen, set_aside, possibly, near
def _value(self, r):
return self.schemes[r.scheme_id]["benefit"].get("value_inr", 0)
# --- 2. Prepare + human approval ------------------------------------
def prepare(self, chosen):
return [{"scheme_id": r.scheme_id, "person": r.person, "raja": self.schemes[r.scheme_id]["raja"]}
for r in chosen]
def approve(self, drafts, approvals):
"""approvals: person -> method ('face', 'voice+otp', 'face@helper-tablet') or None.
Each adult approves her or his own applications; a guardian approves for a minor."""
signed = []
for d in drafts:
method = approvals.get(d["person"])
if method:
d["approval"] = method
signed.append(d)
return signed
# --- 3. File --------------------------------------------------------
def submit(self, signed):
out = []
for d in signed:
scheme = self.schemes[d["scheme_id"]]
sources = {c["_source"] for c in scheme["criteria"]}
self.ledger.delegate(self.mandate.mandate_id, d["raja"], sources)
out.append(Envelope(self.id, d["raja"], "APPLY",
{"scheme_id": d["scheme_id"], "person": d["person"], "approval": d["approval"],
"evidence": {}}, self.mandate.mandate_id))
return out
# --- 4. Respond to everyone -------------------------------------------
def handle(self, env: Envelope):
return getattr(self, "on_" + env.kind.lower())(env)
def _to(self, recipient, kind, body):
return Envelope(self.id, recipient, kind, body, self.mandate.mandate_id)
def on_ack(self, env):
b = env.body
self.claims[b["ref"]] = {"scheme_id": b["scheme_id"], "person": b["person"], "raja": env.sender,
"status": "filed"}
return []
def on_clarify(self, env):
b = env.body
if b["need"] == "permission":
self.tell("need_permission", detail="; ".join(b["detail"]))
return []
# Ask the other side's agent before ever asking the family.
return [self._to(self.peers["college"], "DOC_REQUEST", b)]
def on_doc(self, env):
b = env.body
raja = self.claims[b["ref"]]["raja"]
self.tell("doc_fetched", doc=b["doc"]["type"], issuer=b["doc"]["issuer"])
return [self._to(raja, "EVIDENCE", {"ref": b["ref"], "need": b["need"], "doc": b["doc"]})]
def on_offer(self, env):
b = env.body
ok_days = self.constraints.get("available_days")
options = [o for o in b["options"] if not ok_days or o["day"] in ok_days]
if not options:
self.tell("no_suitable_option")
return []
pick = options[0] # earliest that suits the family
self.tell("slot_chosen", when=f"{pick['date']} ({pick['day']}) at {pick['place']}")
return [self._to(env.sender, "CHOOSE", {"ref": b["ref"], "option": pick})]
def on_decision(self, env):
b = env.body
c = self.claims[b["ref"]]
c["status"] = b["outcome"]
name = self.schemes[c["scheme_id"]]["name"]
if b["outcome"] == "approved":
self.tell("approved", scheme=name, person=c["person"])
elif b["outcome"] == "rejected":
self.tell("rejected", scheme=name, person=c["person"], why="; ".join(b["reasons"]))
return [self._to(env.sender, "APPEAL", {"ref": b["ref"], "grounds": "family disputes the record"})]
return []
def on_payment(self, env):
b = env.body
c = self.claims[b["ref"]]
name = self.schemes[c["scheme_id"]]["name"]
if b["status"] == "credited":
c["status"] = "credited"
self.tell("credited", scheme=name, person=c["person"], detail=b["detail"])
return []
c["status"] = "payment_failed"
cp = f"DARPG/E/2026/{next(_cp)}"
self.tell("payment_failed", scheme=name, person=c["person"], detail=b["detail"], cpgrams=cp)
return [self._to(env.sender, "GRIEVANCE", {"ref": b["ref"], "cpgrams": cp, "issue": b["detail"]}),
self._to(self.peers["bank"], "SERVICE_REQUEST",
{"service": "reactivate_account", "person": c["person"], "ref": b["ref"]})]
def on_service_done(self, env):
ref = env.body["ref"]
self.tell("account_fixed", person=env.body["person"])
return [self._to(self.claims[ref]["raja"], "STATUS", {"ref": ref, "retry_payment": True})]
_________________________________________________________________________________________
praja_raja/raja.py
"""RAJA -- a government agent, one per scheme-owning department.
Guards its scheme's rules and budget. It never takes the citizen's word for a
fact: it re-checks every criterion at source, under the consent the citizen
delegated to it. It may ask for evidence, offer options the rules allow, and
decide -- always with the rule trace. It has no way to ask a citizen to drop
a claim (see protocol.RAJA_MAY_SEND).
"""
from __future__ import annotations
import itertools
from .protocol import Envelope
from .rules import evaluate
_ref = itertools.count(70210, 37)
class Raja:
def __init__(self, raja_id, schemes, registries, bank, slots=None):
self.id = raja_id
self.schemes = {k: v for k, v in schemes.items() if v["raja"] == raja_id}
self.reg = registries
self.bank = bank # PFMS / DBT rail
self.slots = slots or [] # e.g. ALIMCO camp dates
self.cases: dict[str, dict] = {} # ref -> case
self.officer_queue: list[str] = [] # humans handle appeals and exceptions
def _out(self, env, kind, body):
return Envelope(self.id, env.sender, kind, body, env.mandate_id)
# ------------------------------------------------------------------
def handle(self, env: Envelope):
return getattr(self, "on_" + env.kind.lower())(env)
def on_apply(self, env):
b = env.body
scheme = self.schemes[b["scheme_id"]]
ref = f"{self.id.split(':')[1][:6].rstrip('-').upper()}-26-{next(_ref)}"
case = {"ref": ref, "scheme": scheme, "person": b["person"], "evidence": dict(b.get("evidence", {})),
"state": "received", "env": env, "slot": None}
self.cases[ref] = case
out = [self._out(env, "ACK", {"ref": ref, "scheme_id": scheme["id"], "person": b["person"]})]
# 1. Signed by the beneficiary herself?
if not b.get("approval"):
case["state"] = "rejected"
return out + [self._out(env, "DECISION", {"ref": ref, "outcome": "rejected",
"reasons": ["Not approved by the beneficiary"]})]
return out + self._progress(case)
def _progress(self, case):
env, scheme, person = case["env"], case["scheme"], case["person"]
# 2. Facts, checked at source under delegated consent. PRAJA's claims are never used.
result = evaluate(scheme, person,
lambda p, pred, arg: self.reg.ask(self.id, env.mandate_id, p, pred, arg))
if result.status == "not_eligible":
case["state"] = "rejected"
return [self._out(env, "DECISION", {"ref": case["ref"], "outcome": "rejected", "reasons": result.reasons()})]
if result.status == "possibly":
return [self._out(env, "CLARIFY", {"ref": case["ref"], "need": "permission",
"detail": [r for r in result.reasons() if r.startswith("??")]})]
# 3. Evidence the rules require -- any accepted form will do.
for ev in scheme.get("evidence", []):
if case["evidence"].get(ev["need"], {}).get("type") not in ev["accepts"]:
case["state"] = "awaiting_evidence"
return [self._out(env, "CLARIFY", {"ref": case["ref"], "need": ev["need"], "accepts": ev["accepts"]})]
# 4. Options the rules leave open: offered, never imposed.
if "camp_slot" in scheme.get("negotiable", []) and not case["slot"]:
case["state"] = "awaiting_choice"
return [self._out(env, "OFFER", {"ref": case["ref"], "what": "camp_slot", "options": self.slots})]
# 5. Decide, with the full trace.
case["state"] = "approved"
out = [self._out(env, "DECISION", {"ref": case["ref"], "outcome": "approved",
"reasons": result.reasons(), "slot": case["slot"]})]
return out + self._pay(case)
def _pay(self, case):
scheme = case["scheme"]
if scheme["benefit"]["kind"] == "in_kind":
return []
ok, detail = self.bank.credit(case["person"], scheme["benefit"].get("value_inr", 0), case["ref"])
case["state"] = "paid" if ok else "payment_failed"
return [self._out(case["env"], "PAYMENT", {"ref": case["ref"], "status": "credited" if ok else "failed",
"detail": detail})]
def on_evidence(self, env):
case = self.cases[env.body["ref"]]
case["evidence"][env.body["need"]] = env.body["doc"]
return self._progress(case)
def on_choose(self, env):
case = self.cases[env.body["ref"]]
if env.body["option"] not in self.slots:
return [self._out(env, "OFFER", {"ref": case["ref"], "what": "camp_slot", "options": self.slots})]
case["slot"] = env.body["option"]
return self._progress(case)
def on_grievance(self, env):
# Logged with the grievance system; nothing is decided by the agent alone.
self.officer_queue.append(f"grievance {env.body['cpgrams']} on {env.body['ref']}")
return []
def on_status(self, env):
case = self.cases[env.body["ref"]]
if env.body.get("retry_payment") and case["state"] == "payment_failed":
return self._pay(case)
return [self._out(env, "DECISION", {"ref": case["ref"], "outcome": case["state"], "reasons": []})]
def on_appeal(self, env):
self.officer_queue.append(f"appeal on {env.body['ref']}: {env.body['grounds']}")
return []
________________________________________________________________________________________
praja_raja/protocol.py
"""PRAJA <-> RAJA message protocol.
Every message between agents travels in an Envelope. The protocol fixes
WHICH kinds of message each side may send. This is how the two ground rules
are enforced in code rather than by good intentions:
* RAJA (government agent) has no message kind for asking a citizen to drop
or reduce a claim. It can only acknowledge, ask for evidence, offer
choices, decide (with reasons) and report payment.
* PRAJA (citizen agent) has no message kind for asserting facts.
Applications carry consent and evidence references only; RAJA checks the
facts itself, at source. A PRAJA therefore cannot misrepresent.
"""
from __future__ import annotations
import hashlib
import hmac
import itertools
import json
from dataclasses import dataclass, field, asdict
from typing import Any
# --- What each side is allowed to say --------------------------------------
PRAJA_MAY_SEND = {
"APPLY", # apply for a scheme (consent token + evidence refs, no facts)
"EVIDENCE", # reply to a CLARIFY request with a document reference
"CHOOSE", # accept one of the options RAJA offered (slot, variant)
"APPEAL", # contest a rejection; goes to a human officer
"GRIEVANCE", # complaint about delay or failure (CPGRAMS)
"STATUS", # ask where an application stands
}
RAJA_MAY_SEND = {
"ACK", # application received, reference number
"CLARIFY", # evidence missing; lists every kind of evidence it accepts
"OFFER", # options inside the rules (camp slots, delivery mode)
"DECISION", # approved / rejected, always with the rule trace
"PAYMENT", # DBT credited / failed, with reason
}
PEER_KINDS = {"DOC_REQUEST", "DOC", "SERVICE_REQUEST", "SERVICE_DONE"}
_clock = itertools.count(1)
_ids = itertools.count(1)
@dataclass
class Envelope:
sender: str
recipient: str
kind: str
body: dict[str, Any]
mandate_id: str | None = None
msg_id: str = field(default_factory=lambda: f"m{next(_ids):05d}")
t: int = field(default_factory=lambda: next(_clock)) # logical clock
signature: str = ""
def payload(self) -> bytes:
d = asdict(self)
d.pop("signature")
return json.dumps(d, sort_keys=True, ensure_ascii=False).encode()
def sign(self, key: bytes) -> "Envelope":
self.signature = hmac.new(key, self.payload(), hashlib.sha256).hexdigest()
return self
def verify(self, key: bytes) -> bool:
good = hmac.new(key, self.payload(), hashlib.sha256).hexdigest()
return hmac.compare_digest(good, self.signature)
class ProtocolViolation(Exception):
pass
def check_allowed(sender_role: str, kind: str) -> None:
allowed = {"praja": PRAJA_MAY_SEND, "raja": RAJA_MAY_SEND, "peer": PEER_KINDS}[sender_role]
if kind not in allowed and kind not in PEER_KINDS:
raise ProtocolViolation(f"a {sender_role.upper()} agent may not send '{kind}'")
_________________________________________________________________________________________
praja_raja/rules.py
"""Scheme-as-Code: a deterministic rules engine.
Each scheme publishes its eligibility as a list of criteria (schemes.json).
Both PRAJA and RAJA run the SAME engine on the SAME published rules, so they
can never disagree about what the rules say -- only about facts, and facts
are settled at source by the registries.
Every evaluation returns a trace: one line per criterion, met / not met /
unknown, in plain words. That trace is what the citizen hears as "why".
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from pathlib import Path
SCHEMES_FILE = Path(__file__).with_name("schemes.json")
def load_schemes(source_of=None) -> dict[str, dict]:
"""Load the published rules. `source_of(predicate)` tags each criterion
with the registry that answers it (used only for plain-language notes)."""
data = json.loads(SCHEMES_FILE.read_text(encoding="utf-8"))
for s in data["schemes"]:
for c in s["criteria"]:
c["_source"] = source_of(c["predicate"]) if source_of else "source"
return {s["id"]: s for s in data["schemes"]}
@dataclass
class Result:
scheme_id: str
person: str
status: str # "eligible" | "not_eligible" | "possibly"
trace: list[tuple[str, str, str]] # (criterion label, met|not_met|unknown, note)
def reasons(self) -> list[str]:
mark = {"met": "OK ", "not_met": "NO ", "unknown": "?? ", "skipped": "-- "}
return [f"{mark[s]} {label}{(' - ' + note) if note else ''}" for label, s, note in self.trace]
def evaluate(scheme: dict, person: str, ask) -> Result:
"""`ask(person, predicate, arg)` returns True / False / None (no record) / "no-consent"."""
trace, any_unknown, any_fail = [], False, False
for c in scheme["criteria"]:
if any_fail:
# Privacy by design: once a rule fails, ask no further questions about this person.
trace.append((c["label"], "skipped", "not asked"))
continue
ans = ask(person, c["predicate"], c.get("arg"))
if ans is True:
trace.append((c["label"], "met", ""))
elif ans == "no-consent":
trace.append((c["label"], "unknown", f"needs permission to check {c['_source']}"))
any_unknown = True
else:
note = c.get("fail_note", "") if ans is False else "no record found"
trace.append((c["label"], "not_met", note))
any_fail = True
status = "not_eligible" if any_fail else "possibly" if any_unknown else "eligible"
return Result(scheme["id"], person, status, trace)
___________________________________________________________________________________________
praja_raja/schemes.json
{ "_note": "Illustrative Scheme-as-Code entries for the demo. Real criteria, amounts and owners would be published and versioned by each ministry.",
"version": "2026.3-demo",
"schemes": [
{
"id": "NFBS",
"name": "National Family Benefit Scheme",
"raja": "RAJA:UP-SocialWelfare",
"benefit": {"kind": "one_time", "value_inr": 20000},
"criteria": [
{"label": "Spouse (primary breadwinner) has died", "predicate": "spouse_deceased"},
{"label": "Breadwinner was aged 18-59", "predicate": "spouse_age_at_death_between", "arg": [18, 59]},
{"label": "Household is BPL / SECC-deprived", "predicate": "secc_deprived"}
],
"evidence": []
},
{
"id": "IGNWPS",
"name": "Widow Pension (NSAP)",
"raja": "RAJA:UP-SocialWelfare",
"benefit": {"kind": "monthly", "value_inr": 1000},
"exclusive_group": "state-pension",
"criteria": [
{"label": "Is a widow", "predicate": "spouse_deceased"},
{"label": "Aged 40 or more", "predicate": "age_gte", "arg": 40},
{"label": "Aged 79 or less", "predicate": "age_lte", "arg": 79},
{"label": "Household is BPL / SECC-deprived", "predicate": "secc_deprived"}
],
"evidence": []
},
{
"id": "IGNOAPS",
"name": "Old Age Pension (NSAP)",
"raja": "RAJA:UP-SocialWelfare",
"benefit": {"kind": "monthly", "value_inr": 800},
"exclusive_group": "state-pension",
"criteria": [
{"label": "Aged 60 or more", "predicate": "age_gte", "arg": 60},
{"label": "Household is BPL / SECC-deprived", "predicate": "secc_deprived"}
],
"evidence": []
},
{
"id": "PMS-SC",
"name": "Post-Matric Scholarship for SC Students",
"raja": "RAJA:MoSJE-NSP",
"benefit": {"kind": "annual", "value_inr": 0, "note": "fees + maintenance allowance"},
"criteria": [
{"label": "Belongs to a Scheduled Caste", "predicate": "caste_is", "arg": "SC"},
{"label": "Passed Class 12", "predicate": "has_document", "arg": "class12_marksheet"},
{"label": "Family income up to Rs 2.5 lakh", "predicate": "income_below", "arg": 250000},
{"label": "Aged 17 or more", "predicate": "age_gte", "arg": 17}
],
"evidence": [
{"need": "proof of current enrolment", "accepts": ["college_admission_receipt", "aishe_enrolment_record", "bonafide_certificate"]}
]
},
{
"id": "ADIP",
"name": "ADIP - Assistive Aids for Divyangjan",
"raja": "RAJA:DEPwD-ALIMCO",
"benefit": {"kind": "in_kind", "value_inr": 0, "note": "calipers + crutches"},
"negotiable": ["camp_slot"],
"criteria": [
{"label": "Disability of 40% or more (UDID)", "predicate": "disability_pct_gte", "arg": 40},
{"label": "Income within limit", "predicate": "income_below", "arg": 250000},
{"label": "No same aid in the last 3 years", "predicate": "no_aid_last_3y"}
],
"evidence": []
},
{
"id": "PRE-SWD",
"name": "Pre-Matric Scholarship for Students with Disabilities",
"raja": "RAJA:DEPwD-NSP",
"benefit": {"kind": "annual", "value_inr": 0, "note": "annual scholarship"},
"criteria": [
{"label": "Studying in Class 9 or 10", "predicate": "in_class", "arg": [9, 10]},
{"label": "Disability of 40% or more (UDID)", "predicate": "disability_pct_gte", "arg": 40},
{"label": "Family income up to Rs 2.5 lakh", "predicate": "income_below", "arg": 250000}
],
"evidence": []
},
{
"id": "APY",
"name": "Atal Pension Yojana",
"raja": "RAJA:PFRDA-APY",
"benefit": {"kind": "contributory", "value_inr": 0},
"inform_only": true,
"criteria": [
{"label": "Aged 18 or more", "predicate": "age_gte", "arg": 18},
{"label": "Aged 40 or less at entry", "predicate": "age_lte", "arg": 40, "fail_note": "entry closes at 40"}
],
"evidence": []
}
]
}
_________________________________________________________________________________________________
praja_raja/consent.py
"""Consent ledger: the citizen-visible, revocable record of every data use.
A Mandate is what the citizen grants her PRAJA: which sources it may ask,
for what purpose, until when, and how she proved it was her (face, voice+OTP).
When PRAJA applies to a scheme it DELEGATES a narrower slice of that mandate
to the scheme's RAJA, so the department can verify facts at source itself.
Every read, delegation and revocation is written to the log she can see.
"""
from __future__ import annotations
import itertools
from dataclasses import dataclass, field
_mid = itertools.count(7731)
@dataclass
class Mandate:
citizen: str
sources: set[str]
purpose: str
days: int
method: str # "face" | "voice+otp" | "face@helper-tablet"
helper: str | None = None # Anganwadi worker etc., logged, never the consenter
mandate_id: str = field(default_factory=lambda: f"K-{next(_mid)}")
delegations: dict[str, set[str]] = field(default_factory=dict) # raja_id -> sources
revoked: set[str] = field(default_factory=set)
class ConsentLedger:
def __init__(self) -> None:
self.mandates: dict[str, Mandate] = {}
self.log: list[tuple[str, str, str, str]] = [] # (citizen, actor, action, detail)
# --- granting -------------------------------------------------------
def grant(self, citizen, sources, purpose, days=90, method="face", helper=None) -> Mandate:
m = Mandate(citizen, set(sources), purpose, days, method, helper)
self.mandates[m.mandate_id] = m
who = f" (helper: {helper})" if helper else ""
self._log(citizen, "citizen", "GRANT",
f"{m.mandate_id}: {', '.join(sorted(sources))} for {days} days via {method}{who}")
return m
def delegate(self, mandate_id: str, raja_id: str, sources: set[str]) -> None:
m = self.mandates[mandate_id]
sources = sources & (m.sources - m.revoked) # can never widen the mandate
m.delegations[raja_id] = sources
self._log(m.citizen, "PRAJA", "DELEGATE", f"{raja_id} may verify: {', '.join(sorted(sources))}")
def revoke(self, mandate_id: str, source: str) -> None:
m = self.mandates[mandate_id]
m.revoked.add(source)
for s in m.delegations.values():
s.discard(source)
self._log(m.citizen, "citizen", "REVOKE", source)
# --- checking -------------------------------------------------------
def allowed(self, mandate_id: str, actor: str, source: str) -> bool:
m = self.mandates.get(mandate_id)
if not m or source in m.revoked:
return False
if actor.startswith("PRAJA"):
return source in m.sources
return source in m.delegations.get(actor, set())
def record_read(self, mandate_id, actor, source, question, answer) -> None:
m = self.mandates[mandate_id]
self._log(m.citizen, actor, "READ", f"{source}: {question} -> {answer}")
def _log(self, citizen, actor, action, detail) -> None:
self.log.append((citizen, actor, action, detail))
def statement(self, citizen: str) -> list[str]:
"""What the citizen hears in her monthly SMS / sees on her screen."""
return [f"{actor:<22} {action:<9} {detail}" for c, actor, action, detail in self.log if c == citizen]
_______________________________________________________________________________________________
praja_raja/registries.py
"""Stand-ins for the authoritative registries (UIDAI, CRS, UDID, DigiLocker,
NPCI mapper, SECC). In production each is an API of the existing system.
Design rule: a registry answers a yes/no QUESTION about a person; it never
hands over the underlying record. Every question is checked against the
consent ledger first and written to the citizen's log.
"""
from __future__ import annotations
from .consent import ConsentLedger
NO_CONSENT = "no-consent"
# Illustrative data for the demo household (Banda, U.P.).
_DATA = {
"kamla": {"UIDAI": {"age": 61, "state": "UP", "sex": "F"},
"CRS": {"spouse_deceased": True, "spouse_age_at_death": 58},
"SECC": {"deprived": True, "income_lt": 250000},
"NPCI": {"seeded_account": "SBI-2093", "account_active": True},
"PFMS": {"drawing": []}},
"sunita": {"UIDAI": {"age": 18, "state": "UP", "sex": "F"},
"DigiLocker": {"caste": "SC", "class12_pass": True, "docs": {"caste_cert", "class12_marksheet"}},
"SECC": {"deprived": True, "income_lt": 250000},
"NPCI": {"seeded_account": "SBI-4417", "account_active": False}, # will cause a DBT failure
"PFMS": {"drawing": []}},
"ravi": {"UIDAI": {"age": 15, "state": "UP", "sex": "M"},
"UDID": {"disability_pct": 55, "type": "locomotor"},
"UDISE": {"class": 9},
"SECC": {"deprived": True, "income_lt": 250000},
"ALIMCO": {"aid_last_3y": False},
"NPCI": {"seeded_account": "SBI-2093", "account_active": True},
"PFMS": {"drawing": []}},
}
# Each predicate: (source, function(record, arg) -> bool)
_PREDICATES = {
"age_gte": ("UIDAI", lambda r, a: r["age"] >= a),
"age_lte": ("UIDAI", lambda r, a: r["age"] <= a),
"spouse_deceased": ("CRS", lambda r, a: r["spouse_deceased"] is True),
"spouse_age_at_death_between": ("CRS", lambda r, a: a[0] <= r["spouse_age_at_death"] <= a[1]),
"secc_deprived": ("SECC", lambda r, a: r["deprived"] is True),
"income_below": ("SECC", lambda r, a: r["income_lt"] <= a),
"caste_is": ("DigiLocker", lambda r, a: r["caste"] == a),
"has_document": ("DigiLocker", lambda r, a: a in r["docs"]),
"disability_pct_gte": ("UDID", lambda r, a: r["disability_pct"] >= a),
"in_class": ("UDISE", lambda r, a: r["class"] in a),
"no_aid_last_3y": ("ALIMCO", lambda r, a: r["aid_last_3y"] is False),
"has_seeded_account": ("NPCI", lambda r, a: bool(r["seeded_account"])),
"not_drawing": ("PFMS", lambda r, a: a not in r["drawing"]),
}
class Registries:
def __init__(self, ledger: ConsentLedger):
self.ledger = ledger
def source_of(self, predicate: str) -> str:
return _PREDICATES[predicate][0]
def ask(self, actor: str, mandate_id: str, person: str, predicate: str, arg=None):
"""Return True / False, None if no record, or NO_CONSENT if not permitted."""
source, fn = _PREDICATES[predicate]
if not self.ledger.allowed(mandate_id, actor, source):
return NO_CONSENT
rec = _DATA.get(person, {}).get(source)
answer = None if rec is None else bool(fn(rec, arg))
q = f"{person}.{predicate}({arg})" if arg is not None else f"{person}.{predicate}"
self.ledger.record_read(mandate_id, actor, source, q, answer)
return answer
# Used by the bank / PFMS side only, not by agents:
@staticmethod
def account(person):
return _DATA[person]["NPCI"]
@staticmethod
def household():
return list(_DATA)
_________________________________________________________________________________________
praja_raja/peers.py
"""Other parties' agents a PRAJA talks to: the college and the bank.
Also the PFMS/DBT payment rail used by RAJAs. All stand-ins for the demo."""
from __future__ import annotations
from .protocol import Envelope
from .registries import Registries
class CollegeAgent:
"""Govt. P.G. College, Banda -- issues enrolment proof on request."""
id = "PEER:GovtPGCollege-Banda"
def handle(self, env: Envelope):
if env.kind == "DOC_REQUEST":
want = env.body["accepts"]
have = "aishe_enrolment_record" # what this college can issue instantly
if have in want:
return [Envelope(self.id, env.sender, "DOC", {
"ref": env.body["ref"], "need": env.body["need"],
"doc": {"type": have, "id": "AISHE/C-21455/2026/1182", "issuer": self.id}}, env.mandate_id)]
return []
class BankAgent:
"""SBI branch agent -- can reactivate a dormant, Aadhaar-seeded account."""
id = "PEER:SBI-Atarra"
def handle(self, env: Envelope):
if env.kind == "SERVICE_REQUEST" and env.body["service"] == "reactivate_account":
Registries.account(env.body["person"])["account_active"] = True
return [Envelope(self.id, env.sender, "SERVICE_DONE", dict(env.body), env.mandate_id)]
return []
class DBTRail:
"""PFMS-style payment rail."""
def __init__(self):
self.credits = []
def credit(self, person, amount, ref):
acc = Registries.account(person)
if not acc["account_active"]:
return False, f"account {acc['seeded_account']} inactive"
self.credits.append((person, amount, ref))
return True, f"Rs {amount:,} to {acc['seeded_account']}" if amount else f"credited to {acc['seeded_account']}"
_____________________________________________________________________________________________
praja_raja/voice.py
"""Language layer: turns agent decisions into what the family hears / reads.
Here: fixed bilingual templates. In production BHASHINI (speech, translation)
and an LLM would phrase these in the family's own language and dialect --
but only PHRASE them. Every fact and decision comes from the deterministic
agents; the language model is never asked whether someone is eligible.
"""
T = {
"found": ("आपके परिवार के लिए {n} योजनाएँ मिली हैं।", "{n} schemes found for your family."),
"need_permission": ("एक जाँच के लिए आपकी अनुमति चाहिए।", "Your permission is needed for one check: {detail}"),
"doc_fetched": ("ज़रूरी कागज़ हमने सीधे मँगा लिया।", "Fetched {doc} directly from {issuer}."),
"no_suitable_option": ("कोई तारीख़ आपके लिए ठीक नहीं — हम दूसरी माँगेंगे।", "No offered date suits you; asking for others."),
"slot_chosen": ("कैंप की तारीख़ तय हुई।", "Camp booked: {when}."),
"approved": ("आवेदन मंज़ूर हुआ।", "{scheme} approved for {person}."),
"rejected": ("आवेदन नामंज़ूर — हमने अपील कर दी है।", "{scheme} rejected for {person} ({why}). Appeal filed."),
"credited": ("पैसा खाते में आ गया।", "{scheme}: {detail} ({person})."),
"payment_failed": ("पैसा नहीं पहुँचा — शिकायत दर्ज कर दी है।",
"{scheme} payment for {person} failed: {detail}. Complaint {cpgrams} filed; bank asked to fix."),
"account_fixed": ("बैंक खाता फिर से चालू हो गया।", "Bank reactivated {person}'s account; payment retried."),
}
def say(key, **kw):
hi, en = T[key]
return f"{hi.format(**kw)} | {en.format(**kw)}"
___________________________________________________________________________________________
praja_raja/bus.py
"""A minimal message bus between agents.
In production this would be an open, signed agent-to-agent network (in the
spirit of UPI's switch): every PRAJA and RAJA has a registered identity and
key, every message is signed, and every exchange is kept for audit.
Here: an in-memory queue that signs, checks the protocol and keeps a transcript.
"""
from __future__ import annotations
import os
from collections import deque
from .protocol import Envelope, check_allowed
class Bus:
def __init__(self) -> None:
self.agents: dict[str, object] = {}
self.roles: dict[str, str] = {}
self.keys: dict[str, bytes] = {}
self.queue: deque[Envelope] = deque()
self.transcript: list[Envelope] = []
def register(self, agent_id: str, agent, role: str) -> None:
self.agents[agent_id] = agent
self.roles[agent_id] = role
self.keys[agent_id] = os.urandom(32)
def send(self, env: Envelope) -> None:
check_allowed(self.roles[env.sender], env.kind) # raises ProtocolViolation
env.sign(self.keys[env.sender])
self.queue.append(env)
def run(self) -> None:
while self.queue:
env = self.queue.popleft()
assert env.verify(self.keys[env.sender]), "bad signature"
self.transcript.append(env)
for reply in self.agents[env.recipient].handle(env) or []:
self.send(reply)
_____________________________________________________________________________________
tests/test_praja_raja.py
import unittest
from praja_raja.consent import ConsentLedger
from praja_raja.protocol import Envelope, ProtocolViolation, check_allowed
from praja_raja.registries import Registries
from praja_raja.rules import evaluate, load_schemes
def setup(sources=("UIDAI", "CRS", "SECC", "UDID", "UDISE", "DigiLocker", "ALIMCO", "NPCI", "PFMS")):
ledger = ConsentLedger()
reg = Registries(ledger)
schemes = load_schemes(reg.source_of)
m = ledger.grant("household:kamla", set(sources), "test")
ask = lambda p, pred, arg: reg.ask("PRAJA:test", m.mandate_id, p, pred, arg)
return ledger, reg, schemes, m, ask
class Rules(unittest.TestCase):
def test_widow_pension_eligible(self):
*_, schemes, m, ask = setup()
self.assertEqual(evaluate(schemes["IGNWPS"], "kamla", ask).status, "eligible")
def test_same_rules_same_answer(self):
*_, schemes, m, ask = setup()
a = evaluate(schemes["PMS-SC"], "sunita", ask)
b = evaluate(schemes["PMS-SC"], "sunita", ask)
self.assertEqual(a.trace, b.trace)
def test_no_consent_means_possibly_not_rejected(self):
*_, schemes, m, ask = setup(sources=("UIDAI", "SECC", "ALIMCO")) # UDID withheld
self.assertEqual(evaluate(schemes["ADIP"], "ravi", ask).status, "possibly")
def test_stops_asking_after_first_failure(self):
ledger, reg, schemes, m, ask = setup()
before = len(ledger.log)
r = evaluate(schemes["IGNWPS"], "ravi", ask) # Ravi is not a widow
self.assertEqual(r.status, "not_eligible")
self.assertEqual(len(ledger.log) - before, 1) # only one question asked
class Consent(unittest.TestCase):
def test_raja_needs_delegation(self):
ledger, reg, schemes, m, ask = setup()
self.assertEqual(reg.ask("RAJA:DEPwD-ALIMCO", m.mandate_id, "ravi", "disability_pct_gte", 40), "no-consent")
ledger.delegate(m.mandate_id, "RAJA:DEPwD-ALIMCO", {"UDID"})
self.assertTrue(reg.ask("RAJA:DEPwD-ALIMCO", m.mandate_id, "ravi", "disability_pct_gte", 40))
def test_delegation_cannot_widen_mandate(self):
ledger, reg, schemes, m, ask = setup(sources=("UIDAI",))
ledger.delegate(m.mandate_id, "RAJA:X", {"UIDAI", "ITD"})
self.assertEqual(m.delegations["RAJA:X"], {"UIDAI"})
def test_revocation_reaches_rajas(self):
ledger, reg, schemes, m, ask = setup()
ledger.delegate(m.mandate_id, "RAJA:DEPwD-ALIMCO", {"UDID"})
ledger.revoke(m.mandate_id, "UDID")
self.assertEqual(reg.ask("RAJA:DEPwD-ALIMCO", m.mandate_id, "ravi", "disability_pct_gte", 40), "no-consent")
class Protocol(unittest.TestCase):
def test_raja_cannot_ask_to_withdraw(self):
with self.assertRaises(ProtocolViolation):
check_allowed("raja", "WITHDRAW_CLAIM")
def test_praja_cannot_assert_facts(self):
with self.assertRaises(ProtocolViolation):
check_allowed("praja", "ASSERT_FACT")
def test_tampered_message_fails_verification(self):
e = Envelope("PRAJA:a", "RAJA:b", "APPLY", {"scheme_id": "NFBS"}).sign(b"k" * 32)
e.body["scheme_id"] = "IGNWPS"
self.assertFalse(e.verify(b"k" * 32))
if __name__ == "__main__":
unittest.main()
_________________________________________________________________________________________
