PRAJA–RAJA: the code behind the Welfare Doot
Two kinds of agents, one rulebook. PRAJA works for the citizen's household;
RAJA works for a government department. Both read the same published
Scheme-as-Code rules, so they settle facts at source instead of haggling over
eligibility.
13 files, about 1041 lines. Standard-library Python 3.10+, no packages.
Run python demo.py, or open “Demo run” below to see Kamla Devi's household go
from life event to money in the bank. All data is illustrative.
===================================================
PRAJA · citizen's agent · one per household
- Finds every scheme the family qualifies for, asking only yes/no questions
- Gets each adult's own approval before filing anything
- Closes gaps by asking other agents (college, bank) first
- Follows each claim until the money lands; complains if it doesn't
- Cannot assert facts — the protocol has no message for it
RAJA · government agent · one per department
- Re-checks every fact at source, under consent delegated for its scheme only
- Asks for evidence, listing every form it accepts
- Offers options the rules leave open (camp dates)
- Decides with the full rule trace; appeals go to a human officer
- Cannot ask a citizen to drop a claim — the protocol has no message for it
================================================
demo run (output)
===================================================
1. LIFE EVENT -> PRAJA wakes
===================================================
CRS:
death of Ram Prasad (58), Atarra, Banda registered. Household: kamla, sunita, ravi.
Respectful delay:
7 days. Then PRAJA calls Kamla ji (voice, Hindi).
===================================================
2. CONSENT ( Kamla ji : OTP + voice-print on a feature phone )
===================================================
Mandate K-7731: 9 sources, 90 days. Income-tax records NOT shared.
===================================================
3. DISCOVER ( yes/no questions only; zero documents fetched )
===================================================
ELIGIBLE kamla National Family Benefit Scheme
ELIGIBLE kamla Widow Pension (NSAP)
ELIGIBLE sunita Post-Matric Scholarship for SC Students
ELIGIBLE ravi ADIP - Assistive Aids for Divyangjan
ELIGIBLE ravi Pre-Matric Scholarship for Students with Disabilities
SET ASIDE kamla Old Age Pension (NSAP) (can't draw both; the other pension pays more)
INFO sunita Atal Pension Yojana (contributory: explained, family decides)
NO kamla Atal Pension Yojana -- Aged 40 or less at entry (entry closes at 40)
===================================================
4. WHY? ( Sunita's scholarship -- the trace she hears )
===================================================
OK Belongs to a Scheduled Caste
OK Passed Class 12
OK Family income up to Rs 2.5 lakh
OK Aged 17 or more
===================================================
5. APPROVAL ( each adult signs her own; Kamla signs for Ravi, a minor )
===================================================
5 applications signed. Helper: Meena (Anganwadi) lent her tablet camera.
===================================================
6. FILE -> agent-to-agent conversation
===================================================
t1 PRAJA:UP-BND-ATR-0091 -> RAJA:UP-SocialWelfare APPLY {'scheme_id': 'NFBS', 'person': 'kamla', 'approval': 'face@helper-tablet', 'evidence': {}}
t2 PRAJA:UP-BND-ATR-0091 -> RAJA:UP-SocialWelfare APPLY {'scheme_id': 'IGNWPS', 'person': 'kamla', 'approval': 'face@helper-tablet', 'evidence': {}}
t3 PRAJA:UP-BND-ATR-0091 -> RAJA:MoSJE-NSP APPLY {'scheme_id': 'PMS-SC', 'person': 'sunita', 'approval': 'face', 'evidence': {}}
t4 PRAJA:UP-BND-ATR-0091 -> RAJA:DEPwD-ALIMCO APPLY {'scheme_id': 'ADIP', 'person': 'ravi', 'approval': 'face@helper-tablet', 'evidence': {}}
t5 PRAJA:UP-BND-ATR-0091 -> RAJA:DEPwD-NSP APPLY {'scheme_id': 'PRE-SWD', 'person': 'ravi', 'approval': 'face@helper-tablet', 'evidence': {}}
t7 RAJA:UP-SocialWelfare -> PRAJA:UP-BND-ATR-0091 DECISION {'ref': 'UP-SOC-26-70210', 'outcome': 'approved', 'slot': None}
t8 RAJA:UP-SocialWelfare -> PRAJA:UP-BND-ATR-0091 PAYMENT {'ref': 'UP-SOC-26-70210', 'status': 'credited', 'detail': 'Rs 20,000 to SBI-2093'}
t10 RAJA:UP-SocialWelfare -> PRAJA:UP-BND-ATR-0091 DECISION {'ref': 'UP-SOC-26-70247', 'outcome': 'approved', 'slot': None}
t11 RAJA:UP-SocialWelfare -> PRAJA:UP-BND-ATR-0091 PAYMENT {'ref': 'UP-SOC-26-70247', 'status': 'credited', 'detail': 'Rs 1,000 to SBI-2093'}
t13 RAJA:MoSJE-NSP -> PRAJA:UP-BND-ATR-0091 CLARIFY {'ref': 'MOSJE-26-70284', 'need': 'proof of current enrolment', 'accepts': ['college_admission_receipt', 'aishe_enrolment_record', 'bonafide_certificate']}
t15 RAJA:DEPwD-ALIMCO -> PRAJA:UP-BND-ATR-0091 OFFER {'ref': 'DEPWD-26-70321', 'what': 'camp_slot', 'options': [{'date': '2026-11-29', 'day': 'Sat', 'place': 'ALIMCO camp, Banda'}, {'date': '2026-12-03', 'day': 'Thu', 'place': 'ALIMCO camp, Banda'}, {'date': '2026-12-06', 'day': 'Sun', 'place': 'CHC Atarra'}]}
t17 RAJA:DEPwD-NSP -> PRAJA:UP-BND-ATR-0091 DECISION {'ref': 'DEPWD-26-70358', 'outcome': 'approved', 'slot': None}
t18 RAJA:DEPwD-NSP -> PRAJA:UP-BND-ATR-0091 PAYMENT {'ref': 'DEPWD-26-70358', 'status': 'credited', 'detail': 'credited to SBI-2093'}
t19 PRAJA:UP-BND-ATR-0091 -> PEER:GovtPGCollege-Banda DOC_REQUEST {'ref': 'MOSJE-26-70284', 'need': 'proof of current enrolment', 'accepts': ['college_admission_receipt', 'aishe_enrolment_record', 'bonafide_certificate']}
t20 PRAJA:UP-BND-ATR-0091 -> RAJA:DEPwD-ALIMCO CHOOSE {'ref': 'DEPWD-26-70321', 'option': {'date': '2026-12-03', 'day': 'Thu', 'place': 'ALIMCO camp, Banda'}}
t21 PEER:GovtPGCollege-Banda -> PRAJA:UP-BND-ATR-0091 DOC {'ref': 'MOSJE-26-70284', 'need': 'proof of current enrolment', 'doc': {'type': 'aishe_enrolment_record', 'id': 'AISHE/C-21455/2026/1182', 'issuer': 'PEER:GovtPGCollege-Banda'}}
t22 RAJA:DEPwD-ALIMCO -> PRAJA:UP-BND-ATR-0091 DECISION {'ref': 'DEPWD-26-70321', 'outcome': 'approved', 'slot': {'date': '2026-12-03', 'day': 'Thu', 'place': 'ALIMCO camp, Banda'}}
t23 PRAJA:UP-BND-ATR-0091 -> RAJA:MoSJE-NSP EVIDENCE {'ref': 'MOSJE-26-70284', 'need': 'proof of current enrolment', 'doc': {'type': 'aishe_enrolment_record', 'id': 'AISHE/C-21455/2026/1182', 'issuer': 'PEER:GovtPGCollege-Banda'}}
t24 RAJA:MoSJE-NSP -> PRAJA:UP-BND-ATR-0091 DECISION {'ref': 'MOSJE-26-70284', 'outcome': 'approved', 'slot': None}
t25 RAJA:MoSJE-NSP -> PRAJA:UP-BND-ATR-0091 PAYMENT {'ref': 'MOSJE-26-70284', 'status': 'failed', 'detail': 'account SBI-4417 inactive'}
t26 PRAJA:UP-BND-ATR-0091 -> RAJA:MoSJE-NSP GRIEVANCE {'ref': 'MOSJE-26-70284', 'cpgrams': 'DARPG/E/2026/41207', 'issue': 'account SBI-4417 inactive'}
t27 PRAJA:UP-BND-ATR-0091 -> PEER:SBI-Atarra SERVICE_REQUEST {'service': 'reactivate_account', 'person': 'sunita', 'ref': 'MOSJE-26-70284'}
t28 PEER:SBI-Atarra -> PRAJA:UP-BND-ATR-0091 SERVICE_DONE {'service': 'reactivate_account', 'person': 'sunita', 'ref': 'MOSJE-26-70284'}
t29 PRAJA:UP-BND-ATR-0091 -> RAJA:MoSJE-NSP STATUS {'ref': 'MOSJE-26-70284', 'retry_payment': True}
t30 RAJA:MoSJE-NSP -> PRAJA:UP-BND-ATR-0091 PAYMENT {'ref': 'MOSJE-26-70284', 'status': 'credited', 'detail': 'credited to SBI-4417'}
===================================================
7. WHAT THE FAMILY HEARD ( voice call + SMS, Hindi | English )
===================================================
आवेदन मंज़ूर हुआ। | National Family Benefit Scheme approved for kamla.
पैसा खाते में आ गया। | National Family Benefit Scheme: Rs 20,000 to SBI-2093 (kamla).
आवेदन मंज़ूर हुआ। | Widow Pension (NSAP) approved for kamla.
पैसा खाते में आ गया। | Widow Pension (NSAP): Rs 1,000 to SBI-2093 (kamla).
कैंप की तारीख़ तय हुई। | Camp booked: 2026-12-03 (Thu) at ALIMCO camp, Banda.
आवेदन मंज़ूर हुआ। | Pre-Matric Scholarship for Students with Disabilities approved for ravi.
पैसा खाते में आ गया। | Pre-Matric Scholarship for Students with Disabilities: credited to SBI-2093 (ravi).
ज़रूरी कागज़ हमने सीधे मँगा लिया। | Fetched aishe_enrolment_record directly from PEER:GovtPGCollege-Banda.
आवेदन मंज़ूर हुआ। | ADIP - Assistive Aids for Divyangjan approved for ravi.
आवेदन मंज़ूर हुआ। | Post-Matric Scholarship for SC Students approved for sunita.
पैसा नहीं पहुँचा — शिकायत दर्ज कर दी है। | Post-Matric Scholarship for SC Students payment for sunita failed: account SBI-4417 inactive. Complaint DARPG/E/2026/41207 filed; bank asked to fix.
बैंक खाता फिर से चालू हो गया। | Bank reactivated sunita's account; payment retried.
पैसा खाते में आ गया। | Post-Matric Scholarship for SC Students: credited to SBI-4417 (sunita).
===================================================
8. CLAIM STATUS
===================================================
UP-SOC-26-70210 kamla National Family Benefit Scheme credited
UP-SOC-26-70247 kamla Widow Pension (NSAP) credited
MOSJE-26-70284 sunita Post-Matric Scholarship for SC Students credited
DEPWD-26-70321 ravi ADIP - Assistive Aids for Divyangjan approved
DEPWD-26-70358 ravi Pre-Matric Scholarship for Students with Disabilities
credited
Officer queue at RAJA:MoSJE-NSP :
grievance DARPG/E/2026/41207 on MOSJE-26-70284
===================================================
9. GUARDRAILS ( enforced by the protocol, not by goodwill )
===================================================
Refused: a RAJA agent may not send 'WITHDRAW_CLAIM'
Refused: a PRAJA agent may not send 'ASSERT_FACT'
===================================================
10. CONSENT LOG ( what Kamla ji can hear or read -- and revoke )
===================================================
Citizen GRANT K-7731:
ALIMCO, CRS, DigiLocker, NPCI, PFMS, SECC, UDID, UDISE, UIDAI for 90 days via
voice+otp
PRAJA:UP-BND-ATR-0091 READ CRS: kamla.spouse_deceased -> True
PRAJA:UP-BND-ATR-0091 READ CRS: kamla.spouse_age_at_death_between([18, 59]) -> True
PRAJA:UP-BND-ATR-0091 READ SECC: kamla.secc_deprived -> True
PRAJA:UP-BND-ATR-0091 READ CRS: sunita.spouse_deceased -> None
PRAJA:UP-BND-ATR-0091 READ CRS: ravi.spouse_deceased -> None
PRAJA:UP-BND-ATR-0091 READ CRS: kamla.spouse_deceased -> True
PRAJA:UP-BND-ATR-0091 READ UIDAI: kamla.age_gte(40) -> True
PRAJA:UP-BND-ATR-0091 READ UIDAI: kamla.age_lte(79) -> True
PRAJA:UP-BND-ATR-0091 READ SECC: kamla.secc_deprived -> True
PRAJA:UP-BND-ATR-0091 READ CRS: sunita.spouse_deceased -> None
PRAJA:UP-BND-ATR-0091 READ CRS: ravi.spouse_deceased -> None
... 66 entries in all
Kamla ji revokes UDID. Can the ADIP RAJA still check Ravi's disability? -> no-consent
_____________________________________________________________________________
demo.py
"""Kamla Devi's household, end to end.
Run: python demo.py
Needs only Python 3.10+, no packages.
"""
from praja_raja.bus import Bus
from praja_raja.consent import ConsentLedger
from praja_raja.peers import BankAgent, CollegeAgent, DBTRail
from praja_raja.praja import Praja
from praja_raja.protocol import Envelope, ProtocolViolation
from praja_raja.raja import Raja
from praja_raja.registries import Registries
from praja_raja.rules import load_schemes
def h(title):
print(f"\n{'=' * 78}\n {title}\n{'=' * 78}")
ledger = ConsentLedger()
reg = Registries(ledger)
schemes = load_schemes(reg.source_of)
bank, bus = DBTRail(), Bus()
# --- Government side: one RAJA per scheme-owning department --------------
rajas = {rid: Raja(rid, schemes, reg, bank, slots=[
{"date": "2026-11-29", "day": "Sat", "place": "ALIMCO camp, Banda"},
{"date": "2026-12-03", "day": "Thu", "place": "ALIMCO camp, Banda"},
{"date": "2026-12-06", "day": "Sun", "place": "CHC Atarra"}])
for rid in sorted({s["raja"] for s in schemes.values()})}
for rid, r in rajas.items():
bus.register(rid, r, "raja")
college, sbi = CollegeAgent(), BankAgent()
bus.register(college.id, college, "peer")
bus.register(sbi.id, sbi, "peer")
h("1. LIFE EVENT -> PRAJA wakes")
print("CRS: death of Ram Prasad (58), Atarra, Banda registered. Household: kamla, sunita, ravi.")
print("Respectful delay: 7 days. Then PRAJA calls Kamla ji (voice, Hindi).")
h("2. CONSENT (Kamla ji: OTP + voice-print on a feature phone)")
mandate = ledger.grant("household:kamla", {"UIDAI", "CRS", "SECC", "UDID", "UDISE", "DigiLocker", "ALIMCO",
"NPCI", "PFMS"}, purpose="find and apply for welfare schemes",
days=90, method="voice+otp")
print(f"Mandate {mandate.mandate_id}: {len(mandate.sources)} sources, 90 days. Income-tax records NOT shared.")
praja = Praja("UP-BND-ATR-0091", ["kamla", "sunita", "ravi"], "kamla", schemes, reg, ledger, mandate,
constraints={"available_days": {"Thu", "Sun"}}) # Kamla works Saturdays
praja.peers = {"college": college.id, "bank": sbi.id}
bus.register(praja.id, praja, "praja")
h("3. DISCOVER (yes/no questions only; zero documents fetched)")
chosen, set_aside, possibly, near = praja.discover()
for r in chosen:
print(f" ELIGIBLE {r.person:<7} {schemes[r.scheme_id]['name']}")
for r in set_aside:
print(f" SET ASIDE {r.person:<7} {schemes[r.scheme_id]['name']} (can't draw both; the other pension pays more)")
for r in possibly:
print(f" POSSIBLY {r.person:<7} {schemes[r.scheme_id]['name']}")
for r in praja.inform_only:
print(f" INFO {r.person:<7} {schemes[r.scheme_id]['name']} (contributory: explained, family decides)")
for r in near:
why = [t for t in r.trace if t[1] == 'not_met'][0]
print(f" NO {r.person:<7} {schemes[r.scheme_id]['name']} -- {why[0]} ({why[2]})")
h("4. WHY? (Sunita's scholarship -- the trace she hears)")
r = next(x for x in chosen if x.scheme_id == "PMS-SC")
print("\n".join(" " + line for line in r.reasons()))
h("5. APPROVAL (each adult signs her own; Kamla signs for Ravi, a minor)")
drafts = praja.prepare(chosen)
signed = praja.approve(drafts, {"kamla": "face@helper-tablet", "sunita": "face", "ravi": "face@helper-tablet"})
print(f" {len(signed)} applications signed. Helper: Meena (Anganwadi) lent her tablet camera.")
h("6. FILE -> agent-to-agent conversation")
for env in praja.submit(signed):
bus.send(env)
bus.run()
for e in bus.transcript:
if e.kind in ("ACK",):
continue
body = {k: v for k, v in e.body.items() if k not in ("reasons",)}
print(f" t{e.t:<3} {e.sender:<26} -> {e.recipient:<26} {e.kind:<15} {body}")
h("7. WHAT THE FAMILY HEARD (voice call + SMS, Hindi | English)")
for line in praja.outbox:
print(" " + line)
h("8. CLAIM STATUS")
for ref, c in praja.claims.items():
print(f" {ref:<16} {c['person']:<7} {schemes[c['scheme_id']]['name']:<52} {c['status']}")
for rid, r in rajas.items():
for item in r.officer_queue:
print(f" Officer queue at {rid}: {item}")
h("9. GUARDRAILS (enforced by the protocol, not by goodwill)")
try:
bus.send(Envelope("RAJA:UP-SocialWelfare", praja.id, "WITHDRAW_CLAIM", {"ref": "x"}))
except ProtocolViolation as e:
print(f" Refused: {e}")
try:
bus.send(Envelope(praja.id, "RAJA:MoSJE-NSP", "ASSERT_FACT", {"income": 1}))
except ProtocolViolation as e:
print(f" Refused: {e}")
h("10. CONSENT LOG (what Kamla ji can hear or read -- and revoke)")
for line in ledger.statement("household:kamla")[:12]:
print(" " + line)
print(f" ... {len(ledger.statement('household:kamla'))} entries in all")
ledger.revoke(mandate.mandate_id, "UDID")
print(" Kamla ji revokes UDID. Can the ADIP RAJA still check Ravi's disability? ->",
reg.ask("RAJA:DEPwD-ALIMCO", mandate.mandate_id, "ravi", "disability_pct_gte", 40))
_______________________________________________________________________________
raja_raja/praja.py
"""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()
_________________________________________________________________________________________