#!/usr/bin/env python3 PK\-1wh55 __main__.pyfrom plumb.cli import main raise SystemExit(main()) PK\plumb/PK\3s00plumb/__init__.py"""Plumb — proof that a story is done.""" #: Reported in a manifest's provenance (``ch2-11``) so a reader can tell which build observed a #: fact. Static pre-1.0 and deliberately not read from package metadata: this must work from a #: source checkout, where no metadata exists. __version__ = "0.3.4" PK\plumb/adapters/PK=\'ZZplumb/adapters/__init__.py"""Language-bound adapters. Each produces a manifest from a test run and nothing more.""" PK\plumb/adapters/java/PK\WԄnn"plumb/adapters/java/plumb-java.jarPK \ META-INF/PK\META-INF/MANIFEST.MFMLK-. K-*ϳR03r.JM,IMuR0233VpIMLPKrJM86PK \plumb/PK\plumb/Manifest$Citation.classU]SP=H֔BU6|"*E0<(IMn? 3vrܛ6۳g}kGч~Ʊ~=N!dZXfv##" 4FyTo]1S4!Lbΐ,idA qTyCddHCRFBwu!C#R2wU\Ӑ~%ia!,vl^\7Y了ILQ֊v3(!snj 1柛FeX<\׷^On6+ۥ'Lo1:E\;1 T6xt*|͔Ec%r>ۮPqaбuXm hYa<`)x(%Q+ TYll4B!%ׂϷZH虁 an3Dt˳L)PW4}#l0I^&)2:C4|QSB3b1VPY Ƃ@h" AӁny}Xy5o%G6B5A%Cmەڤ&igO>ȌgxϦ4vvOrw=u+l€ c7 Dٲ+ 4U\p1 *!ۨ57qgl%Ln+ !]7Ħpʶp-"`'C藚կm(]'~a ^0ӅSR:`,443ٴo3o=cw0m Eo#-Ҳ6FN+ Q(fiSPI &HIhg |aS$X^4͝a"HYWǸOѾAUAcMAӁT4F'9cQa^^G KTxxC*XKl PKM;@PK\plumb/Manifest$Score.classSNQ]N/ԢriJŻ jhRXƒ>GԙSI$&<~qϥҁ3kgw;fØ =~wDW:=i5XF i נ3SUjM)!kzN%zE*< )%1Ҋ"kT[lcq7akGU14IL m-vL=UgjlVA|˴&e gw,:)l+{U(U">6&*х2m9pfږ=,uIUrSoÙH\~MƠcic3h擨g+x' œ ҝ!?FLPz\v|KA1hx+ܮct,_.~6MO)?y /ԋW /80Vh btctimm$8 -yH1o3U f?#.zD|s㘥4H8͝p uI=]޹ˈeT^.૪ޢ@eDnez_ |E(` 0/kS.#MAJmZ&D%At#ĕzR)2[g%zHLEM03T3mX{ K`z$WʍH驘PY\å-P﫫"0S=+5\ח鳫GߕI섙[-4TnhZEkRT鼪G\28LD.F 4a!q<'U BÕXp&ͽ}~'1nb Rp'&34m4)5㉮ޙ4WJ1ɱL`ʰ2QJ3dܰhO iGOrIJC#'qGT RDF`FU U( X/0dzzUAvl(;o~%Ucjؤ!̰=&܋2naזWUk*R[ҾL"P"btT*rY,Uë@ao0+-qR3t(T 5B:M;}o´t$;FjK$moƵAOFolV os*φhc ViTʫ<`0VGsǫ̃rPO $}\)8(fFō>ǣÒ U|44ϣyPC&bUsxbekS (Fe8/hx_kl֓NSn/iv'؇!|UuWm4nw$ҶӲXZ97-|7[,YוHuuMVf*BG5@ *6 ]6O祿5We ~(zb4db2tXAlʧ϶EڐY?iw>mZ`j.m;>Y Rj~$br],#v#bԣ[QcOƐhOȯ?2s+dbY|ug-A4)]eY:oªmM*ūn{s'kETO[ٞpѾBE5JvsNmIF䔽w982= :(DPSli꬟y,v?F1^^MQ_)(&Ǭ6kn]sKhK{}([QNq,dG)XrGecA O[6zֆX|Gb$җw&3nș>2HcXp+̚YUëdxg<d+$[Ǫ(0/<)."dP&;9U`hqQ[(ϧ~WyKh3VD8$jiQ|6NZR| @!+Q3c +u֜ VC]WhvڶV1<9! 4@ԕ:U-@s2L_׈Xb@Ű^sQq !zs<{GBόcruZ@*#;ӗ'}Ǹ)mG^znLF tISP͊X ^ePڻwh-a5%tZV]p仑ߙ9Yek(1(-;DI=¤M.1.5_6GIBuH~ TSGe8 f1# ̭9 Qs֣(: %\`rbsj!'𷂿a*3C΢~R@MʆiS =9A^Ujp f1erqR=eXsv֒NTxlK}_-Q52f*с~z\@鎻ַrF 5H3Ћ?xMP@4k'7XOX%J pPKo PK\plumb/Mutants$Collector.classXy|Td7@ @ @2!\ID IcH&3,,֥uV(nmۺ`U-.ڪ}Jsdi/{}{w{<T xJtֆtʊ~xQhȄP!6$cDnH‘ YѨP(j Gé 5 ovZDڳ&LQbb&0NaҰcJ(p[,T([3(J40aҶg&&8#I;0nM(tMLTo+ԮaiF64zሕ ǢH&ki&cBq< Wn3{a 3MB5s+БJ]n!P,NC S%<#$ʞxjv,BǛ8'*`1]vjaumVԸõrw xl?e 316"pcq LNu:eB, ;ND%Ӏmb?MV oynb"t1WbQ9<Ȇ5Cq:=Ssdƫ уO&vJ.h$8h=XD;1Ra1E.5q.a A Ţ)+,E5635gZ6]cxjj\cZu9ܸ"{xVX΂c[-$}>hC&>Yl]b;Ul~3qRRq;aEٷ`[M܆fq82VF;Vr$mE;M|B+L%ZIJYԀR6q>-yK)jm^ KJϙqdo S+3܋+üD\9Ź78޶+ey[T1Xs6p96D*i{Ñ[oTtlwE-b͊;kit,%ǜ7ʦ53Byi!0׮\YղU6u*Q뭭\Ѱ=EhS_+f)? h݉.Cw>)٧8qƓ}zNujv5aȏM##f8ؼ*,8/l|dX}cيX,œŊ;$}jG2VTʧݨYSU5֧(ϨvlZWǫ u"/{$AѝvPjʒ_F2NfdgxN׍fۋ PhV$?g1)9HefSyixǤC̣G5C@~, ?q85Pj{e}˾Om|*GY7T9M䬓B Q| DI/LNNEyy *XQj3+C0wɈR.J*+d7s+ͽἢ Ƨ o£8Nað\)]|+6YǪ Gn X7 :ct¦ %C^ 6[xkOR$zh栘ݐB\fRZw )/eTt@3 lxrN<2E ۩ƹToUԛa/ΕP0yE`R{1V[WѤza\PQI"qq]7\=:6J*jR]Kp6&7snqޅPxX- 9:F(e ˗q]/$l"8.$l0ܪ}?{=@a^^p_֒1ȼ -³}*S ;pqJp nj?NY)>*7E!w'eUro>6L+/*/ }c>{amY͝{ *y9?w;y.)2-b^k!=<wI?%7[07/ rŅ~pƫdB)#$ '^;gU-us0[:A6T{+Ɣ`tU !ju,\uxpdya! S 5m4)n8^h Gqy<ܥ#N^mkN0'OԻ>Nl[#d"> S&_"] ,5bQkkvɃ0^x&/=<7rsy'vfNjL (5KnEj9G?k3sZ7BN1Y-找dYJ5I-UPPK|Ţ PK\plumb/Mutants$Mutant.classUmOP~1;7@[L|( a@#nX0'M$q1?eN& bqc/0ԃR{(X#zK?NK?`(Ja@!1\4"h' #1DZQS$\$PJt2HY_ǸOS~B  Y(' 4=:ʚ P",vD3o[c<GOϰ#kg PKX4PK\plumb/Mutants$Site.classTYSA&$7B)& /X$1%Hnܝ'*SVGYl[3+8BPVA!yha=g.)3tds;SA\5 jv_ Ė)8C̮rC [1]zKG>å[E = i Vʌ^c *2$lyCŰtHQN*0N,݈Ȑ͖_ݴM[8KEh(Y6} 4#-M'? @h"½sh=ϭk:` ?:u5$Cx#vD}^ZMO]:뱁v'=UdyGT]e2@=+%ɍ26mC6fl$c+Yі0R@`2xIF`NSdj,J"0[M&G_'bЗ][^N@`w«KF7zȾ .b>ZrgjZ١k"*wްj:xh/N=*s/@8>ɜ>ɡU)ݐ״@JP*cZ%t:>s-zD҆`y dQ"Xj)4%\'C>YIvajJ!ZL2'a8(P9HY;ML ^ˆ O+ X2/+(LH1HK{tNg yAi4[LiӇcUC1#ZM^`qXzXMDžxj`q$kb.̚:U9L]U}"SN~-%?i o$*)}1X)*NeZoےQ8:4d^H@GNtнKLÌ_we :Vb[ ")6?w擜^+CZ"Fi.PjWuj&'㜼YX̄tU;)~6)tb $˘8i+#lCGU]M-vӺmpz7ntAz/ȒI#P_xO RՊ.Gaooh ۂllN~@HtG|b#T/8K>QM_S5\xo%m,SlꜤ?TyۺZ;l0Q:LHI[&BjkAܿͿiluu3)'*7o"l5+R4l~mPecPGFeSC.;O8}G֠Sx/m;j>VVy= 4>9#@ZA rwRTPxf6?- ґNKQet`PQyh GS?<7UYNmatM)ߪi&JU5GA!PVB DT0[̗E5BH· j᫦ԋi{Քp4ߖARIq;b%Z4mo"~f1ka`\-7mȄ+ѨeeRiK\"Ib,*E7y-H2Vt4s_O&ELBQ*tâ1jE2$Q-2Q9yD]ˆ$$(WȾX/H먌(_p'1Cc_XzroSm?GB#bz zkNAGG4p9CbiBqH5ǩ:޺ 8E6\BO`h/è/`Md kq)?^\[̯1lZ~îh^5Pzc;*X[I;"~؆Eh؎N@&vb?MEP g%lPa2(Fzے&%p I8RZ7հڬ9jσqſ&aD|8n6@i OYD{k&@e+8$a^9,zS}!{zx޼^/y$Om=~= 6Sc8NHDHA&"]`Pg}ڦ[쵹O,S)#LxKc#Dx"Ex O+)SLx28Oqj9?Lmy!mXNJ/7b5a-D99rkE4r}a*s<[ATKt%!*xb5DX#|bPKvye vPK\plumb/Mutate$Plan.classV[Se~6 M*m6ZE LڄOؒ/֟ᅷ8rG9>{rX3zy&uf)iH`\`ܪ-gD\oW`|vl*0VL : ^Y)K- d|zWCygt1M1ݮm_\C-} .kZꝀKԦ052. rk:]uz\&*7RF>Yu'_q )ok.m(^ eK~߳ڴLYy*0ѰKHC&Twʣ_37ZeG $_IGm5NoHnYՑJz #Ѿs\CYR&_,'AXI5R99m8 \<'`UGw% deSQ &;t:e{|Y(~!`=nFMlmg))h_aC& \93PvGU'Z^Gv-g-źy/}ؕ.=-OG\fqRJ^CCڿrE 2GpX8}(cb0!} BθyQJ_\Aw.T +{X 4_f~ǜ?#=":c_|ge}8LWb&8XU 6"\1* B5q# .:rL+ڨt(4<y"&OFm0ЧX ׻t\6E3X2Q KF,!:+F?6v07PK{HPK\plumb/Mutate.classZ `Tdd`L DHL :d^d&ֶvZ7pim*nCEߺԪjm]Wj[L&{{9sw<ٸ & loׇM+7n`8_,(t>?Ep؄ea Pwc}["|E"+LXR$]6'H}w0dׯ{*I5iW156TxEz֬h0icrH c3q/N༑ H`aHFUM)OT^b杊L PEDX<Uk+JjPkgs3{7<5sWq8T'Y\~?W43&ŇJ_JgLo".|m0ûXdž15!7Ni&,,"AYi)bz N>.{uAJ0;ۦahL8FW.ȝ^fi>#)23ڨD<o-t,z`̎++,xQmѨgk0FVc1BPGwvUCbtbYX9sW0nGt:䜹-q28x pa&rHhȱ^h #'UU(9JSfJ1lX*Mvl[mRC.u*W5AGe2}b9`yn͝p؆UEUJC"n!mT LWt;,E>vcrO`)X Υ0V3;R^W&·O |-|W2GYf驂%UGGޖA(GQWY>.%zc2}§iMX'Wqc"H(zp 5a9iyN֊u H".$X՛Hv7OWޗ^;hdlo"^7`3[QFo_@nEY3aQf{C?Έyf̞}m^i}^$B=vԮy1o*ޘMq n5q7:[fzn5 &mՊԠ/[؃жZXE7<Ҫ)Z {锁DTGEӲincbޜ,GK tr Fg ͢T7,U؟ !$ L,T Ñ5^[!qXVW4)27-| 59p7([xh_vM;ӴAUNY|Wت~L#5jr])-n8∷-PaP5#!_Tu'/.TP+¯i\Hj/ r%O9/&~c]&W -%@(h1A]Sք3ΗgՈ ` ٛcJF7dJ&"}Eiu?gA隨m8p5Uzto^Xxd 2'BS tMs=3L̹&UTE,1T ET(E8,)Afg!yv,)`Dn^1f4 FaJ[xS&0GI :9t}4 H))˟՗TH%e *C5Fjz層B~cig* ]2}au%^aSbU}LhQ^M9ђJKwVs]Z)ְr`)U#¹52RcIh`l)X07ag̶^Ncmܙ[+c{1 SNaxBeWS4:ЫF0P!ВEr: &eVTN%}ֲw]yKpc pKJaJKv:%%K5XE%UNK :ԠKP~%tq|XEGH}ӔeW\tٱXH'RziKjF^rZ~%t돪c<HyhM9Q%,Y{JU/n[MM\LvT SQ$wM%_=*k &ƮD4ʚ^5熃hnZپEJ똤Z;-YrՄX(: _ne fhd''w7vw,WE`􅭴o%`TH)(~7er0)CB`DC1*9dCT̹Vg`*JQQrsO%Wh1yܔ',<ԶR$9Q(KRr|\ThOml))<˾C托s^ojyfZ3:ʈD{>zK:aK.mL2C j7˖Z~CS>jTĿr--XC٣Vy5VEJԿXW |g-y]u/Z/udJS9@/#)PY5~}e[%kt][4 yH;l l1e҂|c拕/Ma<>]shHÚ<$}G^\VMMc^fkz)Cys12ku=׊̗j+ ^BS0׻P緌ƢcPڦ bMǗHSR2gTOY޺^!nc2^fY"[g>/wΩoj4ajb޺5ْ;_L\C~+͢ uQ^HV#j1ˈl ne{>=WݿL-9ܱLr%MA_n@Woܛ wi\*8nlRMc?ga q<x>zky6&R]?gD3Nә(L jjĄ߃({k1*yTYf@ frYB1ØIȴE\FQsn!Tփp)<8 6_m3C}G $apŦGm釖ƒh] Nc+8W`zl y)2z؎+p>u+ꓭ)?);\ݻ>)7M''+޵'\ҹ=,q{].rs/yP($ zEo.usOzBHRm> {t ӈƀi/)o^[COoDa|[u!هq. QocN%hEKiWx$M`~y-qL"z-:tET/?7̏>Fy\ QZUfq$5]a*+w8.Lʯ_^fT{XuY<V}h˦i&Mj$ޚN˭5)!K!k4D—ue򪑔?+y<¹e 2= Lٽcnrv*?9\VѲ.enr*PNsHcW. r_L]zkrS),+w&iLQ{`9rU/́5 ;i:e+Fw?ί3eo,dU/af#M/.VO!uT?KUcT0T t!)`%nuܒ/PqSHӍ#u< cij=BW~3Ɉm3 lUߵ+՚C_׵N_7i K_m}dl`,"4<gd~qq< PKRi2PK\plumb/Observer.class[ |T;3yĀ!a) @I@Pm$83AZVP*Z *!.V[֥syMf$~#7{ιx}OX~恂CS 9?^y# Gi* nH;V-G;]hp ]010+dFd!1#[Z1I2LFHX} h[ ZBo-ܼIjݏ&#( e8nCF Z#`8AUQ:fh~W# |I,fJmHCw/Z46"P`1|s>WF p>f]j8%.ٙ QaXBldEwh iXրK&Ri8MgLf[xNT\kTOM>{FeXNp 'uLW-oxhҰRǙXfiى:] @m`d/XXU~i?Aq=i& 4zFG?PRl 4h$ 61-9շ;ݡUHl`@Ȩ*m:Ȯ ijDaX+ױAf`eŚ.,萎@{.bjqI/GDCTG ݄1H(060fm*Jw6݋Ꚗ/XYԋ-PE:.%li!W %$ILu9p0hvʍ+صM/\ /* Wצj.B ShAt\x,Ȉ?fAJgs7iqK":FcSL k:ocS9qc; :l;Ѯ`'g3,I͋m e!je3Úޣc/zYiѴABP)yOD0ih MB)4HԈ5uGPp0!YZWJ6d3t7x?⠓&MXVw%!!SryHbث:~b7bD; zMsI3nxQ?%%^+o l.7LW"o`eDt( o(kx0d_^?#$fDPrss9.f]ħfl}R0? l6EIRx_?Ol?*Τ,@4?dg8|1#f<:s'i}k}3|ΉB֒vj.r?䪆ElUj C!& ;1NJj_ww+IoU/(G#M'7JLHFf2,g!K2anNm6!ݔDZExD/KyTHX%e_ :K>}6CXbP'f 8^nX!Q :xvN 77"/ZPfV fRٵK4QI4u Mϱޞ`u7TN}=o}}?)xJ~ϡ=4S/4؜%B./J;Ij^O@>3<:B O 0u4ZbhH;-19CwAքr}/t: 5Mճ1R됬Ds+[\{~Cӕ[rQ[S̤J&̤l1\W#asF=!C#,}9UWNR<]qs:v [=jt5NL΅p.gӎ~&v蘼HT&?*urλp$]vC[Wɺ*UeBC(,d2Vp* t+Tr[cFl1Yp[̳jyuV_bsGcDkj{RI[+O1g47ݫfj]H|vGmsy(Oad-” &}Prt ؘ܊QpQ$Z{Dp\M2}zhbt"4]A<(ғ[F*A5jZ.J[&<$*!BHpt\inu&*ťWj K{=cSlRQ9 kj]ګٕÐWj-n}d*sSY-NV<.dt<9_1 c,fU#NE阧8Э_ ~cY}ֺ9.]wYg]Qvy *)'%1_m,+;ES=uGwVsS3mbv o&P%'4-"냮&4$$Asֆ[ yQوc%gͺP` u:1Jaf!ƌk" Y@T G$.M&L [/_LZ,;/6rM:L6վbWN8 2kFo="R-.ta2%ܖ)i1O %>L*DX}%uͣq#rM\ȫZuMnjtR[V6A(y[㪻tu6sIʮ<*&e([Օw3?p [Ũꃺ"O}?l?E+""*R4ZV4e~nB;@Oc e]HYS Y[=mk_Jnd>TDrw2Ve?tu]ootc5YG+\{1C:Y]Ò ֍Z[ G1-{qiqPujW Q,w`Nkϕ[v8fX;fJJC(OE`rwl/>_NvyÄ8ek4q`Њ+&<9xRfnǒ>J 3iItV6aoB6.0ᇌ4_ް(/o~LOsе/,Q[63W}~om>s+}qCw}o; djOӃ[;1U??}z>ىQ2~hv8fv91aydA)u;^& s#ס7Iu;5 .8Dhi 6Q#jl!?...p ]K*\N WJ*zW;>õ*שrlSp:7SqjMjnVpZ[Ml/Pn [5JA3oBy3&Z70TvOPry~I1JrpKuͿ]@w:7V]i.zRٔ,{f{u7J|cќ3,ϰBb<oo〶-6Xwq|{ͿB؄Ø rQ!CN-nq&r0f)RgSSCP%Q!*YK"IڼШV*y oTRxj3N8U<W9WSnGb e"9Cn$5 W J`*cؔ*ĝVe8(s +٠L]Q^C ]N-aS:3C񽴆wL: oEa7aҝSJjaB\Qȏ_eEU4;؝$e2 y8ż씋S.]u(D/&hG_Q"A4kE>n$V6I`c ^h].HW:-bt.:y'p"ф9<E7~Kq_?b^ykW9pf3:?8E~MaaLx91ǖ31E=G}8|[9bgܶyLm73TMjP:~nߐ0T=UItwbrĴSqYje;tŷ[FKfgZȟ3/O5e2!L [ɋ?{R$0.;+J33ddӭ;y97ӌ imOeN1>+ًqosØ'd ޮ ށk lw +VOw{Տ.S3K}8CO%Q8>b~̉SDqqy|'%Hd8Nω叨~ve3/"a7Tp.$d!_Nտ'IxɁ*rbtҰh\[jqP&bSxwIO*lNzE3.eǻpudWwOw/ Hb*@5ccdh^z- K;6ˊüq27XfCv<@o7rg h`,̕tCћ4umH me~<6NqxVݎuS EݕY9ϓB^{;<{0QQ (I=G c}\&еGy$VQnIVVdpeAvVp/G1T,FM@ 4u\evd8K,ÍTۨR%xUxq>450/pl_aXè .)\U(#}ǶvW [k*,ʸӪPRDB@ʹLG*/S5>^"?PK wPK\plumb/PlumbListener.classWitƖdy0Ɓ)ll `)NP@IXك3b4&mB4m!M6!mos8q7#ے-;q8Gystx4,mP~Z5*^);ɨ`yV6IQ*|7M4;mcqb*\ɑb+߂ Mlhrm)ѵA=“5V~x5zSOP)#EǍl{ާ vd5LW"3k5ׇT|aŷNHa0I\ $C8S*>٩&o[Lu) 6.fU|YӖ㼠Vq-wlb*AFM&K:f۔n{}aS\=ni)pA ۦ'|>DZ gU<*J&⢊%mi>>|^m535i@r~n|(;:7hie pjҨfF,wv2t6t! UKq7a${?r T^7z#lfJn!Mhd[]Zp};FeJ`W!p W yI&ѧ'NbcɗP{AeH+K%J >RobBؓ ~{0~&-q9~_/^ZN -_TSo,ܿ7PK?PK\plumb/PlumbProbe.class1OAqi /#-9{fYPQƷ@`y3_ޛ7g\5QC=F@J>8OLOJ-KQ]"z-𞊔Q/+;OdɹT6sS?>JM0Xʤ;e+3bUYQ٥ig vW׃iPq0"B{[c\_:\_#]RkPKjWyPK\plumb/PlumbRunner.classX xGG&Si*$$4s4bĤi46vʉZPr7PEvH(7rP(*U蛕d˲|jwslgy`]C.CʨN*pw ͪ[ӈj^x ˘ >k_S>L0!`̥L n3K):3fۡI1,L'pxFԌ5nx[m`ELjWS>dԕ*9vAm6MIq W؊(JsCZn)TTȘO@_dXRFxs@51Le\V w1\{@_`?Ur4@\ͤ3h mCMnakt[ Igk zV)lW=5_=3$EDM"PQ2rI02+I#X =K\KVMn(n0, ]/-9$ڄGLADF U)20LXR1xg.n;i#S(ە͂$Ž$a񵙙Q1e85!g(IPTH2>SrL@Z$|at!iMk >+s2>/9I'?eI'B˃Tڗe _q5˗ }b k 1jj&!ƅ>5zvlq"יcH BFBL <<uň =p&Q4;2TR2 6"ߑ]<, Q miP*G}!%HƏV=aҷЛM~NDz,KtiD5utqւ5ZF2h]&Rڥ&yQSi+{ZQK&w+J޸aSIތ6B Mj;YhIDㅫU}YڄZ4^kθw*,g1к% Xī \?8p٬ay8ZQd2dvğS n^07 Rj/RLNw/"tũR5Ft}$J0*8:SDLF0vK[ z50ЯADjCTj,wmu5♺BQ*AbK 5-1?&T :B ⟃넋! 8C'_e-.@~_#yCS򨋄N;.XN߂ޓh8p {|X:6>9 ~w<ү /I=@RDmV ^,J,Fa-4i@/.ap3ނUnGD BZ(ݭxJgf=W%p%ۂ։lfQt7agTStM&mX I&kZ~ZF׏G6eMC ׇ> cU.i*KɿѱR򍊸!"kTv6wq㾈fs犥Jܨ!i&ֱr<$iIILĸS9&f&s9V4ɨU^/G3DFl1OS(syް0ZWu T(T֔s rћU lV Z/'xLR,_X1rəE,(0EWzeCif4B{`h0C6,@=xB_r~o;;}!5x [M# EEhp p9vBG8Ⱥ#`́ۿNw;KXebe^x>H#4$da#Aҏ tU1:9:1? DŽvYS'*PK}2PK \ META-INF/PK\rJM86+META-INF/MANIFEST.MFPK \plumb/PK\̕;plumb/Manifest$Citation.classPK\M;@plumb/Manifest$Provenance.classPK\C @oplumb/Manifest$Score.classPK\f, l plumb/Manifest.classPK\o plumb/MutantRun.classPK\|Ţ <plumb/Mutants$Collector.classPK\X4Z&plumb/Mutants$Mutant.classPK\B;m)plumb/Mutants$Site.classPK\vye v,plumb/Mutants.classPK\{Hs7plumb/Mutate$Plan.classPK\Ri2W;plumb/Mutate.classPK\Y:VSplumb/Observer.classPK\ wnplumb/PlumbListener$1.classPK\?pplumb/PlumbListener.classPK\jWy!zplumb/PlumbProbe.classPK\*5# R{plumb/PlumbRunner.classPK\)0Dplumb/Proves$Citations.classPK\W6mplumb/Proves.classPK\}2kplumb/RunFile.classPK}PK\plumb/adapters/python/PK=\INC##!plumb/adapters/python/__init__.py"""The Python/pytest adapter. Named for the *language* rather than the runner: the trace is CPython's own, and only the plugin is pytest-shaped. Everything import-specific to Python stops here — what leaves is the neutral manifest (`docs/adapter-contract.md`). """ from plumb.adapters.python.adapter import PlumbAdapter, collect from plumb.adapters.python.archetypes import archetype_of, declare from plumb.adapters.python.surface import ProductionSurface __all__ = ["PlumbAdapter", "collect", "ProductionSurface", "declare", "archetype_of"] PK \W !plumb/adapters/python/__main__.py"""The Python adapter, run out of process: resolved facts in, a manifest out. The same contract the Java adapter has, and it exists for the same reason. Once Plumb is installed as its own artifact, **the interpreter running the tool is not the interpreter running the project** — the adapter has to import the project's code and its test framework, so it must run where those live. That is a subprocess, and a subprocess needs a file rather than an object. Symmetry is the point rather than a tidiness. Both adapters are handed the same file and answer with the same document, so the launcher's job is identical in both directions and a third language is a third reader of this file. Usage: ``python -m plumb.adapters.python `` """ import sys from pathlib import Path from plumb import runfile from plumb.adapters.python import archetypes from plumb.adapters.python.adapter import collect from plumb.adapters.python.mutation import OptIn from plumb.adapters.python.surface import ProductionSurface def main(argv: list[str] | None = None) -> int: args = list(argv if argv is not None else sys.argv[1:]) if len(args) < 2: print("usage: python -m plumb.adapters.python ", file=sys.stderr) return 2 facts = runfile.read(Path(args[0])) out = Path(args[1]) # Declared before the run, so a project whose spec is not executed by this run still has them; # a spec that *is* executed declares over the top, because the spec is the owner (`ch2-7`). for story, kind in (fact[:2] for fact in facts.get(runfile.ARCHETYPE, ()) if len(fact) >= 2): archetypes.declare(story, kind) surface = ProductionSurface.of( *runfile.values(facts, runfile.PRODUCTION), entry_points=runfile.values(facts, runfile.ENTRY_POINT), ) mutation = runfile.values(facts, runfile.MUTATION) == ("true",) # Everything this prints is progress, and the manifest is the product. They go to different # places so a caller reading one is never handed the other (`ch5-2`). marker = runfile.values(facts, runfile.MARKER) manifest = collect(list(runfile.values(facts, runfile.PYTEST_ARG)), surface=surface, marker=marker[0] if marker else "proves", mutation=OptIn(enabled=mutation), report=lambda line: print(line, file=sys.stderr, flush=True)) manifest.write(out) return 0 if __name__ == "__main__": raise SystemExit(main()) PK \t,, plumb/adapters/python/adapter.py"""The Python/pytest adapter: a test run in, a manifest out. This is the **language-bound** half of ``ch0-4``. Everything pytest-shaped stops here — node ids, marks, report phases — and what leaves is the neutral manifest the core reads. The core imports none of this, and importing it from :mod:`plumb.core.gate` or :mod:`plumb.core.board` would collapse the seam that makes a second language a second adapter and nothing else. It reports **facts, not verdicts** (``ch2-5``): what each citing test claimed and how the runner finished it. It never decides proven from unproven, so no adapter can grant or revoke a status. Grounding is computed here because the trace and the entry-point declarations are both language-specific (``ch3-3``, ``ch3-4``); the core reads only the resulting fact. Tracing is scoped to the **call** phase deliberately — a module's imports run at collection, so keeping the window to the test body is what stops import-time frames being mistaken for the test executing production code (``ch3-U4``). """ from dataclasses import dataclass, field import pytest from plumb.adapters.python.archetypes import archetype_of from plumb.adapters.python.grounding import fact from plumb.adapters.python.mutation import RECOVERED, OptIn, fill, restore from plumb.core.manifest import ( FAILED, Provenance, NOT_CHECKED, PASSED, SKIPPED, Citation, Manifest, Story, ) from plumb.adapters.python.surface import ProductionSurface from plumb.adapters.python.trace import tracing #: The default name. It is a **convention**, not a mechanism, so a project that already cites its #: requirements under another name says so in config rather than rewriting every call site #: (`ch9-4`). Adoption cost is measured from the project somebody already has. MARKER = "proves" #: What this adapter is, for the manifest to carry beside its facts (``ch2-11``). The mechanism #: is named because it is what decides whether a fact here is comparable with one from elsewhere: #: ``sys.settrace`` watches the calling thread only, so this adapter's `inert` rests on a #: per-thread call tree and its `dispatched` on a coarse "another thread was alive". ADAPTER_NAME = "plumb-python" MECHANISM = "sys.settrace (per-thread call tree)" def _provenance() -> Provenance: import platform from plumb import __version__ return Provenance( adapter=ADAPTER_NAME, version=__version__, runtime=f"{platform.python_implementation()} {platform.python_version()}", mechanism=MECHANISM, ) #: This hook brackets the test from inside its own frame, so pytest resumes it while the trace #: is still live and it lands in the call tree it is collecting. Naming it here keeps the #: observer out of its own observation. _HOOK = frozenset({"PlumbAdapter.pytest_runtest_call"}) @dataclass class PlumbAdapter: """A pytest plugin that accumulates one manifest per session. Citations are read at collection and outcomes at report, because a test's *claim* and its *result* arrive from pytest at different moments and only the pair is a fact. """ #: what the project declared about itself; an empty surface yields `not-checked` (`ch3-7`) surface: ProductionSurface = field(default_factory=ProductionSurface.of) #: what this project calls a citation (`ch9-4`) marker: str = MARKER #: marker names seen on cited-looking tests that were not ours, so an empty board can say #: *why* it is empty instead of leaving the reader to guess foreign: set = field(default_factory=set) #: node id -> the (story, depth, ref) triples that test claims claims: dict[str, list[tuple[str, str, str | None]]] = field(default_factory=dict) #: node id -> normalized runner result outcomes: dict[str, str] = field(default_factory=dict) #: node id -> the grounding fact derived from its trace grounding: dict[str, str] = field(default_factory=dict) #: node id -> did it also run production code no entry point reached (`ch3-5`) reached_outside: dict[str, bool] = field(default_factory=dict) #: node id -> the production frames it ran *under a declared entry point*, which is what #: mutation is scoped to (`ch4-3`). Kept from the same trace grounding was derived from, so #: mutation adds no second observation and no map of story to code. wired: dict[str, tuple[str, ...]] = field(default_factory=dict) # ---- pytest hooks ------------------------------------------------------------------- def pytest_configure(self, config): # noqa: D401 — hook """Declare the marker, because a plugin that adds one and does not register it makes pytest warn `Unknown pytest.mark.proves - is this a typo?` on every citation — the tool telling the user their claim might be a mistake while quietly relying on it. Under ``--strict-markers``, which real projects turn on, it is a failure rather than a warning.""" config.addinivalue_line( "markers", f"{self.marker}(id, ..., depth, ref): cite the stories this test proves, at a depth " f'— e.g. @pytest.mark.{self.marker}("APP-1", depth="unit"). `ref` is optional and is ' "your own external key, which Plumb carries and never parses", ) @pytest.hookimpl(hookwrapper=True) def pytest_runtest_call(self, item): """Trace the test body, and only the body. A hookwrapper so the trace brackets exactly the call phase: setup and teardown are somebody else's frames, and collection-time imports have already happened. Uncited tests are not traced at all — tracing costs real time, and a test that claims nothing produces no fact anyone reads. """ if item.nodeid not in self.claims: yield return with tracing(self.surface, ignore=_HOOK) as recorder: yield trace = recorder.sealed() self.grounding[item.nodeid] = fact(trace, self.surface) self.reached_outside[item.nodeid] = trace.ran_outside_entry_point self.wired[item.nodeid] = trace.executed_under_entry def pytest_collection_modifyitems(self, items) -> None: for item in items: for mark in item.iter_markers(self.marker): if not mark.args: continue depth = mark.kwargs.get("depth", "") ref = mark.kwargs.get("ref") # EVERY id, not just the first. A project writing several per marker was having # all but one silently discarded, which reads on the board exactly like a story # nobody ever claimed — the one failure this tool exists to prevent. for story in mark.args: self.claims.setdefault(item.nodeid, []).append((story, depth, ref)) self.foreign.update( mark.name for mark in item.own_markers if mark.name != self.marker and mark.args and isinstance(mark.args[0], str)) def pytest_runtest_logreport(self, report) -> None: if report.nodeid not in self.claims: return self.outcomes[report.nodeid] = _worse( self.outcomes.get(report.nodeid), _normalize(report) ) # ---- the product -------------------------------------------------------------------- def manifest(self) -> Manifest: """Group the citations under their opaque story id (``ch2-2``), as facts.""" stories: dict[str, list[Citation]] = {} for node, claims in self.claims.items(): # Never reported on, or reported only for phases that cannot pass (a green setup # whose body never ran) — either way a claim, not proof. result = self.outcomes.get(node) or SKIPPED # A test that never reached its body has no trace, so its fact is `not-checked` # rather than a guess about what it would have touched. grounding = self.grounding.get(node, NOT_CHECKED) for story, depth, ref in claims: stories.setdefault(story, []).append( Citation(test=node, depth=depth, result=result, grounding=grounding, ref=ref, reached_outside_entry=self.reached_outside.get(node)) ) # The archetype enters from the *spec*, not the run — the one thing here read off the # requirement rather than the execution (`ch2-7`). Undeclared means behavioral. return Manifest( stories={ sid: Story(archetype=archetype_of(sid), citations=tuple(cs)) for sid, cs in sorted(stories.items()) }, observer=_provenance(), ) def _normalize(report) -> str | None: """The runner's own signal, mapped without translation loss (``ch2-4``). Only the call phase can report a pass: a test whose body never ran has proved nothing, so a green setup is not an outcome. """ if report.outcome == "failed": return FAILED if report.outcome == "skipped": return SKIPPED return PASSED if report.when == "call" else None #: Worst-wins, so a test that passes its body and then errors in teardown is not a pass. _SEVERITY = {None: 0, PASSED: 1, SKIPPED: 2, FAILED: 3} def _worse(current: str | None, incoming: str | None) -> str | None: return current if _SEVERITY[current] >= _SEVERITY[incoming] else incoming def collect(args: list[str] | None = None, surface: ProductionSurface | None = None, mutation: OptIn | None = None, report=None, marker: str = MARKER) -> Manifest: """Run pytest under the adapter and return the manifest it produced. Both declarations arrive already resolved, because the launcher read them (``plumb.config``). Defaults exist for a caller driving this directly — a test, or a library user — and never for the command line, which always knows what the project declared. Mutation runs **after** the session and only when opted in (``ch4-2``) — it needs the trace and the outcome, and both are only complete once the run is. """ # Before the runner starts, and whether or not mutation is on this time: a mutant left by a # killed run breaks the *test run*, so recovering after it would produce a tidy tree and a # manifest measured against broken code. recovered = restore() if recovered and report: report(RECOVERED.format(path=recovered)) adapter = PlumbAdapter(surface=surface or ProductionSurface.of(), marker=marker) pytest.main(list(args or []), plugins=[adapter]) manifest = adapter.manifest() if not manifest.stories and adapter.foreign and report: # An empty board is the question "why?", and this is the one answer the adapter has that # nobody else does: it saw markers, just not the one it was told to read. report(f"no tests cite `{marker}`, but these markers are in use: " f"{', '.join(sorted(adapter.foreign))}\n" f"If one of them is how this project cites requirements, say so in plumb.toml:\n" f' [python]\n marker = "{sorted(adapter.foreign)[0]}"') return fill(manifest, adapter.wired, adapter.surface, mutation or OptIn(), report=report) PK=\1X::#plumb/adapters/python/archetypes.py"""What kind of claim each story makes, declared by the spec as it runs (``ch1-7``). The archetype is authored **on the story, in the spec** — never on the test, which would let the thing being judged pick its own gate, and never inferred, which would make it guesswork. So the one place it can be collected from is the spec declaring itself. That happens naturally: a spec module's ``story(...)`` calls execute when the runner collects them, so a run that includes the spec has already been told every archetype by the time the manifest is assembled. A run that excludes the spec learns none, and everything is behavioral — which is exactly ``ch1-7``'s documented default, so a spec that never mentions archetypes is unchanged and a narrowed run is degraded rather than wrong (``ch0-7``). This is process-global on purpose. It mirrors the runner's own lifecycle rather than threading a registry through every caller, and :func:`reset` exists so a test can assert on a clean one. """ from plumb.core.manifest import ARCHETYPES, BEHAVIORAL, InvalidManifest _declared: dict[str, str] = {} def declare(story_id: str, archetype: str = BEHAVIORAL) -> None: """Record a story's declared archetype. Called by the spec's claim vocabulary.""" if archetype not in ARCHETYPES: raise InvalidManifest( f"{story_id}: archetype {archetype!r} is not one of {ARCHETYPES} — " "non-functional is not a third kind (ch1-7)" ) _declared[story_id] = archetype def declared() -> dict[str, str]: """Every archetype the spec has declared so far, as a snapshot.""" return dict(_declared) def archetype_of(story_id: str) -> str: """Behavioral is the default, so silence is a valid and complete answer.""" return _declared.get(story_id, BEHAVIORAL) def reset() -> None: _declared.clear() PK\Z*44"plumb/adapters/python/grounding.py"""Chapter 3's fact: did the citing test drive the real system? (``ch3-4``) Deliberately separate from :mod:`plumb.adapters.python.trace`. The trace reports **observations** — which production frames ran, which files were read as data, whether anything ran under a declared entry point. This module turns those into the single ``grounding`` value the manifest carries and the gate reasons over. Keeping them apart is what let the vocabulary be reconsidered without touching the tracer, and the vocabulary is still the unsettled part (``ch3-U4``). The order of questions is fixed by ``ch3-8`` and is not an implementation preference: **inertness is asked first**, because it needs no entry points. A project whose spec is entirely structural declares no production surface at all, and must still get a real fact rather than ``not-checked``. """ from plumb.core.manifest import DISPATCHED, GROUNDED, INERT, NOT_CHECKED, UNGROUNDED from plumb.adapters.python.surface import ProductionSurface from plumb.adapters.python.trace import Trace def fact(trace: Trace, surface: ProductionSurface) -> str: """Derive the grounding fact for one citing test. ``inert`` is not a weaker ``ungrounded``: it is a different answer to a different question. ``ungrounded`` means production code ran and reached no declared entry point — the orphaned-"done" fake. ``inert`` means none ran at all, which is what inspecting an artifact as data looks like, and is the positive proof a structural story needs (``ch1-2-3``). """ if not surface.declares_production_code: # `ch3-8` says inertness needs no *entry points*, and that is true — but it does need # **roots**. With none declared, no frame is ever recognised as production, so an empty # call tree means "nobody said what production code is", not "the test ran none". Those # are different answers and only one of them is `inert`; reporting inert here would hand # every structural story a free pass on a project that declared nothing at all. return NOT_CHECKED if not trace.executed: # `inert` is a positive claim about the whole program, and this tracer is per-thread. With # other threads live, an empty call tree means "nothing ran *here*" — a different and much # weaker statement (`ch3-8`). Saying `inert` anyway would hand a passing grade to a # structural story whose test dispatched its work somewhere unwatched. return INERT if trace.accounted_for_all_threads else DISPATCHED if not surface.declares_entry_points: # Production code ran, but nothing says where the real system begins, so the question # cannot be asked. Surfaced, never silently skipped (`ch3-7`). return NOT_CHECKED # Deliberately NOT degraded to `dispatched`, and the asymmetry with the branch above is the # whole point. This adapter's evidence that work escaped is coarse — "some other thread was # alive" — which is sound for withholding `inert`, a *positive* proof that can grant a # structural story a pass it did not earn. Applied here it would mean any process with a # background thread could never report `ungrounded`, and `ungrounded` is `ch3-5`: orphaned # code cannot ground. Destroying the tool's headline check to fix a narrower error is the # wrong trade. # # The residual gap is real and bounded: a test that runs some production code locally *and* # dispatches its entry-point work elsewhere reads `ungrounded`, which overstates what was # observed. It costs a story its status and can never grant one. Closing it needs a *precise* # dispatch signal — a hand-off actually seen — which this tracer has no way to obtain and an # adapter observing `Executor.submit` would (`ch3-4`). return GROUNDED if trace.ran_under_entry_point else UNGROUNDED PK{\^RR plumb/adapters/python/mutants.py"""Chapter 4's operators: a minimal in-house AST mutation (``ch4-4``). Three source mutations, applied to Python's own parse tree — flip a comparison, negate a branch, swap a return. No PIT, no mutmut, no engine to take a version on: the same principle as the grounding tracer (``ch3-3``), for the same reason. Plumb needs one narrow answer, and a dependency that answers a hundred is a surface it then has to carry. Nothing here runs anything. This module turns source into *mutated source*; deciding whether a citing test notices is :mod:`plumb.adapters.python.mutation`. Keeping them apart is what lets the operator set be argued over — it is a cost/catch tradeoff and openly unsettled (``ch4-U1``) — without touching the part that runs tests. Mutation is **scoped to named functions** (``ch4-3``), never to a whole file. The names are the ones the trace already produced, so what gets mutated is the code the citing test was observed to run, and nothing else. """ import ast from dataclasses import dataclass from typing import Iterator FLIP_COMPARISON = "flip-comparison" NEGATE_BRANCH = "negate-branch" SWAP_RETURN = "swap-return" #: The whole set (``ch4-4``). Small on purpose: these three catch the assert-nothing test, which #: is the residue mutation exists to dent (``ch1-2-U1``). A larger set catches more at more cost, #: and where that line sits is not fixed here (``ch4-U1``). OPERATORS = (FLIP_COMPARISON, NEGATE_BRANCH, SWAP_RETURN) #: Each comparison paired with its negation, so the mutant is a real behavioural change rather #: than a neighbouring one. ``<`` becomes ``>=`` and not ``>``: swapping for a *neighbour* leaves #: the boundary case behaving identically, which is an equivalent mutant on exactly the input a #: test is most likely to use (``ch4-6``). _FLIPPED = { ast.Eq: ast.NotEq, ast.NotEq: ast.Eq, ast.Lt: ast.GtE, ast.GtE: ast.Lt, ast.Gt: ast.LtE, ast.LtE: ast.Gt, ast.In: ast.NotIn, ast.NotIn: ast.In, ast.Is: ast.IsNot, ast.IsNot: ast.Is, } #: The frame name CPython gives module-level code. A test that imports production code during the #: traced window grounds into this, so it is a mutable scope like any other. MODULE_SCOPE = "" #: A nested function and a nested class are separate scope units with their own qualified names, #: so they are mutated only when separately grounded. A lambda is not — it has no frame the trace #: names, so it belongs to the function that encloses it. _OWN_SCOPE = (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef) @dataclass(frozen=True) class Mutant: """One mutated rendering of a module, and where the change was made. ``source`` is the whole module, not a patch: what runs a mutant is a file on disk and a fresh interpreter, so a mutant that cannot be written out is no mutant at all. """ operator: str where: str line: int source: str def mutants(source: str, *, module: str, scope: tuple[str, ...]) -> Iterator[Mutant]: """Every mutant of ``source`` within the named functions, one site at a time. Sites are enumerated from a fresh parse each time rather than by copying a tree, so a mutant can never carry a previous mutant's change — one mutant, one difference, or a survivor means nothing. """ for index, (operator, _, _, where, line) in enumerate(_sites(ast.parse(source), module, scope)): working = ast.parse(source) _, node, at, _, _ = _sites(working, module, scope)[index] _apply(operator, node, at) yield Mutant(operator=operator, where=where, line=line, source=ast.unparse(ast.fix_missing_locations(working))) def site_count(source: str, *, module: str, scope: tuple[str, ...]) -> int: """How many mutants :func:`mutants` would yield, for one parse instead of one per mutant. The count is what a cost estimate is made of, and an estimate that costs as much as the run it is estimating is not an estimate. """ return len(_sites(ast.parse(source), module, scope)) def _sites(tree: ast.Module, module: str, scope: tuple[str, ...]) -> list[tuple]: """``(operator, node, at, where, line)`` for every mutable point, in a stable order. Order matters more than it looks: it is the only thing tying a site found in one parse to the same site in the next, and a mutant applied to the wrong node is a silent false result. """ wanted = frozenset(scope) found = [] for qualname, unit in _scopes(tree): if qualname not in wanted: continue for node in _own_nodes(unit): found.extend( (operator, node, at, f"{module}:{qualname}", node.lineno) for operator, at in _mutable(node) ) return found def _mutable(node: ast.AST) -> Iterator[tuple[str, int]]: """Which operators apply to this node, and at which sub-position.""" if isinstance(node, ast.Compare): yield from ( (FLIP_COMPARISON, at) for at, op in enumerate(node.ops) if type(op) in _FLIPPED ) elif isinstance(node, (ast.If, ast.While)): yield NEGATE_BRANCH, 0 elif isinstance(node, ast.Return): yield SWAP_RETURN, 0 def _apply(operator: str, node: ast.AST, at: int) -> None: if operator == FLIP_COMPARISON: node.ops[at] = _FLIPPED[type(node.ops[at])]() elif operator == NEGATE_BRANCH: node.test = ast.UnaryOp(op=ast.Not(), operand=node.test) else: node.value = _swapped(node.value) def _swapped(value: ast.expr | None) -> ast.expr: """A different value, not merely a missing one. A bare ``return`` and ``return None`` become ``return True`` rather than the reverse: swapping a returned value for ``None`` when it already was ``None`` is the definition of an equivalent mutant, and every one of those costs a full test run to learn nothing (``ch4-U2``). """ if value is None or (isinstance(value, ast.Constant) and value.value is None): return ast.Constant(True) if isinstance(value, ast.Constant) and isinstance(value.value, bool): return ast.Constant(not value.value) return ast.Constant(None) def _scopes(tree: ast.Module) -> Iterator[tuple[str, ast.AST]]: """``(qualified name, node)`` for module-level code and every function under it. The names are built the way CPython builds ``co_qualname`` — ``Class.method``, and ``outer..inner`` for a closure — because the trace reports what the *runtime* called a frame, and a scope this cannot name is a scope that can never be mutated. """ yield MODULE_SCOPE, tree yield from _functions(tree, "") def _functions(node: ast.AST, prefix: str) -> Iterator[tuple[str, ast.AST]]: for child in ast.iter_child_nodes(node): if isinstance(child, ast.ClassDef): yield from _functions(child, f"{prefix}{child.name}.") elif isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)): yield f"{prefix}{child.name}", child yield from _functions(child, f"{prefix}{child.name}..") else: yield from _functions(child, prefix) def _own_nodes(unit: ast.AST) -> Iterator[ast.AST]: """Every node belonging to this scope unit, stopping at anything that owns its own.""" for child in ast.iter_child_nodes(unit): if isinstance(child, _OWN_SCOPE): continue yield child yield from _own_nodes(child) PK3\a::!plumb/adapters/python/mutation.py"""Chapter 4's check: does the story's tests actually *check* the code? (``ch4-1``) Grounding proves the code was **used** (chapter 3). This is the narrower guard on what grounding cannot see — a test that runs the production code and asserts nothing about it, the assertion-meaningfulness residue chapter 1 names and declines to claim (``ch1-2-U1``). Break the code; a test citing the story must notice. **It finds real defects.** Applied to this project it has caught tests that could not fail and code no test reached, on work that a full green suite had already signed off. So the question is never whether to run it — it is **when**, and that is a question about cost: * A mutant costs one **test run**, and a story's mutants are run against **only the tests citing that story** — not the suite (``ch4-3``). That is the whole reason this is affordable at all. * Even so, the multiplier is large. Measured on Plumb itself — a small project whose tests run in fractions of a second — the suite takes ~11s and one full pass takes **45 minutes, 177x that**. Where the tests are slower, so is every one of the thousands of runs. Nothing here makes that cheaper; what it does is let a caller **see the price before paying it** (:func:`plan`), so the decision is made with a number rather than a guess. * Which is why the honest place for it is a story or two at a desk, or a nightly pass — and why it is opt-in (``ch4-2``). Not because it is optional: because it is not a thing you leave on by accident. The result is a **score**, not a verdict (``ch4-5``). A story's tests routinely run far more code than they check, so "did any mutant survive?" is *yes* for almost every honest test and carries almost no information; ``12/40`` and ``39/40`` do not mean the same thing and must not read the same. It stays advisory whatever the numbers say (``ch4-6``) — an equivalent mutant is indistinguishable from a real survivor (``ch4-U2``). What is not done here is limiting the pass to what *changed* since last time. That needs a definite diff and the mechanism is open (``ch4-U3``); it is also the single biggest lever on the cost above, which is why it is worth naming rather than quietly omitting. """ import os import subprocess import sys import time from dataclasses import dataclass, replace from pathlib import Path from typing import Callable from plumb.adapters.python.mutants import Mutant, mutants, site_count from plumb.adapters.python.surface import ProductionSurface from plumb.core.manifest import Manifest, Mutation #: A mutant can make a loop never terminate — negating a ``while`` condition is one of three #: operators — so every run is bounded. A timeout counts as **killed**: the tests did not complete #: against the broken code, which is the opposite of passing over it unmoved. TIMEOUT_SECONDS = 60 @dataclass(frozen=True) class OptIn: """``ch4-2`` — mutation runs when a project asks for it, and not before. Named rather than passed as a bare boolean so the default lives in one place and reads as the decision it is. Config for the same reason grounding takes its surface from config (``ch3-2``): the artifact under test never imports the thing judging it. """ enabled: bool = False @classmethod def from_config(cls, config) -> "OptIn": """Absent means off, which is the whole of ``ch4-2``.""" return cls(enabled=config.mutation) @dataclass(frozen=True) class Target: """One production file, and the functions in it the story's tests drove.""" path: Path module: str qualnames: tuple[str, ...] @dataclass(frozen=True) class Plan: """What one story's mutation pass would run — **before** any of it runs. Separate from doing it because the cost is the decision. A caller that cannot ask "how much" without paying it can only find out by starting, and a run that takes hours is exactly the one nobody should start blind. """ story: str tests: tuple[str, ...] targets: tuple[Target, ...] mutants: int @property def runnable(self) -> bool: return bool(self.mutants and self.tests) def scope(wired: tuple[str, ...], surface: ProductionSurface) -> tuple[Target, ...]: """``ch4-3`` — the grounded code, grouped by the file that holds it. The input is the trace's ``executed_under_entry``: production frames that ran with a declared entry point above them. Scoping to the *whole* call tree instead was measurably wrong — it pulls in every helper a test brushed past, so almost every story reported a survivor and the fact stopped distinguishing anything. A project that declares no entry points therefore gets no mutants and no fact, which is the same answer grounding gives it (``ch3-7``): the question needs a wired system to be asked about, and saying nothing is honest where inventing an answer is not. It reuses the trace rather than adding a second observation, so no story-to-code map is written down — that is the artifact ``ch3-1`` refuses to keep, and mutation is not the place to smuggle one back in. """ plan: dict[Path, tuple[str, set[str]]] = {} for name in wired: path = surface.locate(name) if path is None: continue module, _, qualname = name.partition(":") plan.setdefault(path, (module, set()))[1].add(qualname) return tuple( Target(path=path, module=module, qualnames=tuple(sorted(qualnames))) for path, (module, qualnames) in sorted(plan.items()) ) def plan(story: str, tests: tuple[str, ...], wired_by_test: dict[str, tuple[str, ...]], surface: ProductionSurface) -> Plan: """Cost this story's pass without running any of it. The scope is the **union** across every citing test, and every mutant is then run against **all** of them. A mutant only one of three citing tests would catch is still caught, and ``survived`` comes to mean "no test of this story noticed" rather than "the test I happened to pick did not" — which is what the reader assumes it means anyway. """ targets = scope(tuple(sorted({n for t in tests for n in wired_by_test.get(t, ())})), surface) return Plan( story=story, tests=tests, targets=targets, mutants=sum(site_count(t.path.read_text(), module=t.module, scope=t.qualnames) for t in targets), ) def run(one: Plan, root: Path | str | None = None) -> Mutation | None: """Run every mutant in the plan against every citing test, and score it (``ch4-1``). ``None`` is *not reported* rather than a pass (``ch2-10``): a story whose tests drove nothing wired, or drove code with no mutable site in it, was never asked the question. A score of ``0/0`` would read like an answer. Every mutant runs — there is no stopping at the first survivor. Stopping is what a boolean could afford, and it is why a boolean was cheap: the run ended early precisely when it had learned the least. """ if not one.runnable: return None at = Path(root) if root is not None else Path.cwd() killed = sum( _noticed(mutant, target.path, one.tests, at) for target in one.targets for mutant in mutants(target.path.read_text(), module=target.module, scope=target.qualnames) ) return Mutation(total=one.mutants, killed=killed) def fill(manifest: Manifest, wired_by_test: dict[str, tuple[str, ...]], surface: ProductionSurface, opt_in: OptIn, root: Path | str | None = None, report: Callable[[str], None] | None = None) -> Manifest: """Populate every story's mutation slot — only when the opt-in is on (``ch4-2``, ``ch4-5``). Only **passing** citations are run. A failing test kills every mutant it is handed for the reason it was already failing, so the score would credit a test for not working — the one place mutation could manufacture confidence rather than check it. """ if not opt_in.enabled: return manifest plans = {sid: plan(sid, tuple(sorted({c.test for c in story.citations if c.ran_and_passed})), wired_by_test, surface) for sid, story in manifest.stories.items()} _announce(plans, root, report) # Two stories cited by the same tests over the same code face the same mutants and must get # the same score, so running them twice buys nothing. Stories cluster this way in practice — # one end-to-end test commonly cites five — and at one test run per mutant the saving is the # difference between a pass somebody starts and one they abandon. scored, seen, started = {}, {}, time.perf_counter() for i, (sid, one) in enumerate(sorted(plans.items()), 1): shape = (one.tests, one.targets) if shape not in seen: seen[shape] = run(one, root) scored[sid] = seen[shape] if report and one.runnable: report(f" [{i}/{len(plans)}] {sid}: {scored[sid].killed}/{scored[sid].total} killed " f"({time.perf_counter() - started:.0f}s elapsed)") return replace(manifest, stories={ sid: replace(story, mutation=scored[sid]) for sid, story in manifest.stories.items() }) def _announce(plans: dict[str, Plan], root: Path | str | None, report: Callable[[str], None] | None) -> None: """Say what this will cost before it is spent, in units measured on this machine. A floor rather than a forecast, and labelled as one: it is the count times the cost of starting a test process here, which every mutant pays before running a single assertion. The real figure is higher by however long the citing tests take. Understating is the safe direction for a floor — nobody is talked into a run by a number that turns out to be low. """ if report is None: return total = sum(p.mutants for p in plans.values()) runnable = [p for p in plans.values() if p.runnable] if not total: report("mutation: nothing to mutate — no story drove code under a declared entry point") return floor = _process_floor(runnable[0].tests, Path(root) if root is not None else Path.cwd()) report(f"mutation: {total} mutants over {len(runnable)} stories — at least " f"{_duration(total * floor)} ({floor:.2f}s process floor, measured here), plus however " "long the citing tests take") def _duration(seconds: float) -> str: """Whichever unit a reader can act on. "0 min" is the number that stops being a warning.""" if seconds < 90: return f"{seconds:.0f}s" if seconds < 5400: return f"{seconds / 60:.0f} min" return f"{seconds / 3600:.1f} hours" def _process_floor(tests: tuple[str, ...], root: Path) -> float: """Start a test process and collect one story's tests, without running them. Measured against a real citing-test set rather than an empty run, because collection is where a project's own conftest, plugins and imports get paid, and on a large project that is most of the fixed cost. Timing an empty collection would understate by exactly the amount that grows with the project — the direction that turns a warning into a surprise. """ started = time.perf_counter() _failed((*tests, "--collect-only", "--no-header"), root) return time.perf_counter() - started #: Where the original is parked while a mutant stands in for it. Named and left in the project #: root on purpose: if it is still there, something went wrong and the tree needs putting back. RESCUE = ".plumb-mutation-rescue" def _noticed(mutant: Mutant, path: Path, tests: tuple[str, ...], root: Path) -> bool: """Write the mutant, run the citing tests, put the file back. The original is restored in a ``finally``, which covers every failure Python can raise — but not the process being killed, and mutation runs are long enough that being killed is a normal way for one to end. So the original is also written to disk (:data:`RESCUE`) *before* the mutant goes down, and only removed once the file is whole again. A run that dies mid-mutant leaves that file behind, and the next run refuses to start until it is dealt with. This is not hypothetical: an interrupted run on a working copy silently produced a *measurement* that was wrong by an order of magnitude, and nothing about it looked wrong. In-place mutation buys the thing that matters — the tests run against the project exactly as it is, with its own imports, plugins and configuration — and this is its price, paid openly. """ original = path.read_bytes() rescue = root / RESCUE rescue.write_bytes(str(path).encode() + b"\n" + original) try: path.write_text(mutant.source) return _failed(tests, root) finally: path.write_bytes(original) rescue.unlink(missing_ok=True) class InterruptedRun(RuntimeError): """A previous mutation run died with a mutant still on disk.""" #: What a caller says when it puts a file back, so the two things a reader needs — that a run #: died, and which file it was — arrive together and in the same words wherever it happens. RECOVERED = "mutation: put back {path} — a previous run was interrupted mid-mutant" def restore(root: Path | str | None = None) -> Path | None: """Put back whatever an interrupted run left mutated, and say which file it was. Called **before the tests run**, not before the next mutation pass — which is the ordering the obvious placement gets wrong. A leftover mutant corrupts the *test run* first of all, and by the time a mutation pass wants the file, every fact in the manifest was already measured against broken code. Recovering then puts the tree right and leaves the answer wrong. Automatic rather than an instruction, because the failure this guards against is silent: a mutated source still imports, still runs, and answers every question slightly wrong. """ at = Path(root) if root is not None else Path.cwd() rescue = at / RESCUE if not rescue.is_file(): return None target, _, original = rescue.read_bytes().partition(b"\n") path = Path(target.decode()) path.write_bytes(original) rescue.unlink() return path def _failed(args: tuple[str, ...], root: Path) -> bool: """Run pytest in a fresh interpreter. A subprocess because the mutant has to be *imported*, and this process already holds the original module object.""" try: completed = subprocess.run( [sys.executable, "-m", "pytest", *args, "-x", "-q", "-p", "no:cacheprovider"], cwd=root, capture_output=True, timeout=TIMEOUT_SECONDS, # A cached .pyc keyed to the original would be a mutant that never ran. env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"}, ) except subprocess.TimeoutExpired: return True return completed.returncode != 0 PK3\#S plumb/adapters/python/surface.py"""What the project declares about itself: where its production code lives, and where the wired system begins (``ch3-2``). Deliberately small and stable. This is *not* a per-story map of code — that is the artifact that rots and the reason grounding is computed from a call tree instead. """ from dataclasses import dataclass from pathlib import Path from types import CodeType @dataclass(frozen=True) class ProductionSurface: """The declared surface: production roots, and the entry points within them. ``entry_points`` are ``"module.path:function"`` strings — the form is the adapter's to choose (``ch3-U1``); the core never sees it, only the resulting fact. """ roots: tuple[Path, ...] entry_points: frozenset[str] = frozenset() @classmethod def of(cls, *roots: Path | str, entry_points: tuple[str, ...] = ()) -> "ProductionSurface": return cls( roots=tuple(Path(r).resolve() for r in roots), entry_points=frozenset(entry_points), ) @classmethod def from_config(cls, config) -> "ProductionSurface": """Take the surface from the project's declaration, already read (``ch3-2``). Config rather than a decorator on entry functions, which was the other candidate (``ch3-U1``). A decorator makes production code **import the verifier**, so the artifact under test carries a dependency on the thing judging it — a real adoption cost, and backwards for a tool whose whole claim is that it observes from outside. Config costs the project nothing at runtime and leaves the source untouched. The file is read by the launcher, not here: one reader for every language is what lets a Java project be configured the same way, and an adapter that reads its own config is an adapter that has to agree with all the others about the format. A project declaring nothing gets an empty surface, which is `not-checked` rather than an error: absent input degrades and surfaces (``ch0-7``). """ return cls( roots=config.resolved(*config.production), entry_points=frozenset(config.entry_points), ) @property def declares_production_code(self) -> bool: """False means nothing here can be judged: with no roots, no frame is ever *ours*, so an empty call tree means "we were told nothing", not "the test ran nothing".""" return bool(self.roots) @property def declares_entry_points(self) -> bool: """False means grounding cannot be computed — the `not-checked` case (``ch3-7``).""" return bool(self.entry_points) def owns(self, filename: str) -> bool: """Is this file production code? Path containment, not import machinery.""" if not filename or filename.startswith("<"): return False try: path = Path(filename).resolve() except (OSError, ValueError): return False return any(path.is_relative_to(root) for root in self.roots) def is_entry_point(self, code: CodeType) -> bool: return self.qualify(code) in self.entry_points def locate(self, name: str) -> Path | None: """``pkg.mod:fn`` → ``/…/pkg/mod.py`` — :meth:`qualify` run backwards. It lives beside its inverse because the naming scheme is one decision, not two, and a reader who changes ``qualify`` has to be looking at the thing that undoes it. ``None`` means the name came from outside the declared roots, which :meth:`qualify` renders with a bare stem and cannot be resolved back to a file. """ module = name.partition(":")[0].split(".") for root in self.roots: if module[0] != root.name: continue rest = module[1:] for candidate in (root.joinpath(*rest).with_suffix(".py") if rest else None, root.joinpath(*rest, "__init__.py")): if candidate is not None and candidate.is_file(): return candidate return None def qualify(self, code: CodeType) -> str: """``/…/pkg/mod.py`` + ``fn`` → ``pkg.mod:fn`` — the citable name of a frame.""" path = Path(code.co_filename) # Resolve once: this runs per traced frame, and resolving is a syscall. resolved = path.resolve() for root in self.roots: if resolved.is_relative_to(root): rel = resolved.relative_to(root).with_suffix("") parts = [root.name, *rel.parts] if parts[-1] == "__init__": parts.pop() return f"{'.'.join(parts)}:{code.co_qualname}" return f"{path.stem}:{code.co_qualname}" PK{\ʜ**plumb/adapters/python/trace.py"""Chapter 3's mechanism: a language-native per-test trace (``ch3-3``). The runtime's own tracing around a single test — never a third-party coverage tool. Plumb needs one narrow answer, so it owns a tiny trace and nothing more. The trace records **two** kinds of contact with the production artifact, because the call tree alone cannot tell a structural rule from a test that did nothing: * **executed** frames — production code the test ran, and whether any of it ran under a declared entry point; * **reads** — production files the test opened as *data*, which is what a rule over dependency direction, containment or format actually does. Deriving the grounding fact from this (``ch3-4``) is deliberately *not* here. A probe showed the current four-value vocabulary is unstable for import-touching tests, so the trace exposes the observations and the vocabulary settles separately. Nothing here presumes the answer. """ import sys import threading from contextlib import contextmanager from pathlib import Path from dataclasses import dataclass, field from types import FrameType from typing import Iterator from plumb.adapters.python.surface import ProductionSurface # The tracer's own frames, which must never appear in what it observes. Reading a file fires the # audit hook, and the hook plus the context manager live *here* — inside any project whose # production root contains Plumb. So a test that purely reads an artifact as data, which is the # definition of `inert` (`ch3-8`), recorded `plumb.adapters.python.trace:tracing` as executed and read as # UNGROUNDED. That broke structural stories specifically, and only when Plumb traced itself. # An observer that shows up in its own observation is measuring the wrong thing. _SELF = frozenset({__file__, str(Path(__file__).resolve())}) # An audit hook cannot be removed once installed, so exactly one is added at import and it # consults whichever recorder is active. Installing per-test would leak a hook per test. _active: "_Recorder | None" = None _hook_installed = False def _audit(event: str, args: tuple) -> None: if _active is None or event != "open" or not args: return target = args[0] if isinstance(target, bytes): target = target.decode("utf-8", "replace") if isinstance(target, str): _active.note_read(target) def _install_audit_hook() -> None: global _hook_installed if not _hook_installed: sys.addaudithook(_audit) _hook_installed = True @dataclass(frozen=True) class Trace: """What one test did to the production artifact — observations, not a verdict.""" executed: tuple[str, ...] reads: tuple[str, ...] ran_under_entry_point: bool #: The subset of ``executed`` that ran with a declared entry point above it. Grounding needs #: only the *whether* above; this is the *which*, and it is what mutation is scoped to #: (``ch4-3``). The distinction is the difference between mutating the wired system a test #: drove and mutating every helper it happened to touch on the way past. executed_under_entry: tuple[str, ...] = () #: Did any production frame run with *no* entry point above it? Grounding is a property of #: the whole test (``ch3-1``), so a test that drives an entry point and also exercises #: unreachable code still grounds — this is what makes that visible (``ch3-5``). ran_outside_entry_point: bool = False #: Could this observation account for every thread? ``sys.settrace`` is per-thread, so with #: other threads live an empty call tree means "nothing ran *here*", which is not the same #: claim as "nothing ran" (``ch3-8``). accounted_for_all_threads: bool = True @property def touched_nothing(self) -> bool: """Neither ran nor read production code. `assert True` lands here — and today's spec would call it `inert` and let it prove a structural story.""" return not self.executed and not self.reads @dataclass class _Recorder: """Collects frames via ``sys.settrace`` and reads via the audit hook.""" surface: ProductionSurface executed: list[str] = field(default_factory=list) reads: list[str] = field(default_factory=list) under_entry: list[str] = field(default_factory=list) _entry_depth: int = 0 _under_entry: bool = False _outside_entry: bool = False #: filename -> is this ours. `owns` resolves a path, which is a syscall, and the same #: handful of filenames recur for the entire run. _owned: dict[str, bool] = field(default_factory=dict) #: (filename, qualname) -> its citable name, for the same reason _names: dict[tuple[str, str], str] = field(default_factory=dict) #: set while the recorder is doing its own bookkeeping, so it never observes itself _busy: bool = False _other_threads: bool = False #: qualnames belonging to the machinery that *starts* the trace. A caller that brackets a #: test from inside its own frame — a pytest hookwrapper does exactly this — is resumed #: while tracing is live, so its frame lands in the very call tree it is collecting. _ignore: frozenset[str] = frozenset() def note_read(self, filename: str) -> None: """Record a read, without the recording itself becoming something to record. Python suppresses tracing *inside* the trace function, so frames it creates are safe. The **audit hook is a separate mechanism and gets no such protection**, so the ownership check this makes — which lives in :mod:`plumb.adapters.python.surface` — was being traced as production code the test ran. Excluding this module alone was not enough for that reason. """ self._busy = True try: if self._owns(filename): self.reads.append(filename) finally: self._busy = False def _owns(self, filename: str) -> bool: cached = self._owned.get(filename) if cached is None: cached = self._owned[filename] = self.surface.owns(filename) return cached def __call__(self, frame: FrameType, event: str, arg: object) -> "_Recorder | None": """The global trace hook. **Returning ``None`` is the point**, not an oversight. Whatever this returns becomes the *local* trace function for that frame, receiving a callback for every line it executes. Returning ``self`` unconditionally — which this did — means line-tracing the entire process for the whole traced window. Plumb needs only ``call`` events, which arrive here regardless of what a previous frame returned, plus ``return`` for entry points so their depth window can close. Everything else is answering a question nobody asked, and it cost a 164x slowdown when the traced window happened to contain a nested test session. """ if self._busy: return None code = frame.f_code if event == "call": if (code.co_filename in _SELF or code.co_qualname in self._ignore or not self._owns(code.co_filename)): return None self._enter(code) # Only an entry point needs its own return event. return self if self.surface.is_entry_point(code) else None if event == "return" and self.surface.is_entry_point(code): self._entry_depth -= 1 return None def _name(self, code) -> str: key = (code.co_filename, code.co_qualname) cached = self._names.get(key) if cached is None: cached = self._names[key] = self.surface.qualify(code) return cached def _enter(self, code) -> None: name = self._name(code) self.executed.append(name) if self.surface.is_entry_point(code): self._entry_depth += 1 if self._entry_depth > 0: self._under_entry = True self.under_entry.append(name) else: # Entered with nothing declared above it. The entry point's own frame increments the # depth first, so it is never counted here — only code no entry point reached. self._outside_entry = True def sealed(self) -> Trace: return Trace( executed=tuple(self.executed), reads=tuple(self.reads), ran_under_entry_point=self._under_entry, executed_under_entry=tuple(self.under_entry), ran_outside_entry_point=self._outside_entry, accounted_for_all_threads=not self._other_threads, ) def note_threads(self) -> None: """Sampled at both ends of the window rather than watched continuously. Deliberately conservative and deliberately cheap: a thread that lived entirely between the two samples is missed, and any thread that merely *existed* counts even if it did nothing. Both errors push toward `dispatched`, which costs a story the top status and never grants it — the only direction a wrong guess is allowed to go here. **Uniform across every supported runtime, on purpose.** Python 3.12 added ``threading.settrace_all_threads``, which would let this watch the pool threads and answer `grounded` or `inert` properly instead of withholding. Using it where available would mean the same test on the same code yielding a *different grounding fact* depending on the interpreter — into a manifest with no field naming which runtime observed it, so a board could mix measurements from instruments it cannot tell apart. Better answers on some runtimes are not worth facts that silently disagree across them. When the manifest carries provenance, this is the first thing to revisit. """ if threading.active_count() > 1: self._other_threads = True @contextmanager def tracing(surface: ProductionSurface, ignore: frozenset[str] = frozenset()) -> Iterator["_Recorder"]: """Trace one test. Yields the recorder; read ``.sealed()`` after the block. Nests by saving and restoring whatever tracer was already installed, so a host that traces for its own reasons is left as it was found. """ global _active _install_audit_hook() recorder = _Recorder(surface, _ignore=ignore) previous_tracer, previous_active = sys.gettrace(), _active _active = recorder recorder.note_threads() sys.settrace(recorder) try: yield recorder finally: sys.settrace(previous_tracer) recorder.note_threads() _active = previous_active def trace_call(surface: ProductionSurface, fn, *args, **kwargs) -> Trace: """Run ``fn`` under the trace and return what it touched.""" with tracing(surface) as recorder: fn(*args, **kwargs) return recorder.sealed() PKG\GG plumb/cli.py"""``plumb`` — the command line: the board, and friction capture. ``plumb board`` runs the adapter over a test run and emits the derived result (``ch5-1``). ``plumb friction`` captures what rubbed (``ch6-1``, ``ch6-10``). """ # Everything a user reads — help text, the board, a refusal — is written for someone who has # never seen this source. Story ids, chapter numbers and the arguments behind a design belong in # the code and the spec, where the reader can follow them; in a terminal they are noise the reader # cannot act on. What earns its place here is what the command does and what it will touch. FRICTION_DESCRIPTION = """Capture what rubbed while applying the method. Records are written to your own store and nowhere else. Nothing is transmitted unless this installation is enrolled, and anything you file can be withdrawn. """ import argparse import os import sys from pathlib import Path from plumb.friction import INAPPLICABLE, FrictionRecord, InvalidRecord, excerpt from plumb.store import FrictionStore, mint_id, utc_now from plumb.transmit import drain, send_one from plumb.triggers import instruction DEFAULT_STORE = Path( os.environ.get("PLUMB_FRICTION_DIR") or Path(os.environ.get("XDG_DATA_HOME", Path.home() / ".local" / "share")) / "plumb" / "friction" ) def _store(args) -> FrictionStore: return FrictionStore(args.store or DEFAULT_STORE) def cmd_file(args) -> int: location, body = INAPPLICABLE, INAPPLICABLE if args.code: path, _, span = args.code.rpartition(":") start, _, end = span.partition("-") try: location, body = excerpt(path, int(start), int(end or start), root=args.root) except (ValueError, OSError) as bad: print(f"--code: {bad}", file=sys.stderr) return 2 try: record = FrictionRecord( id=args.id or mint_id(args.observation), observed_at=utc_now(), source="self", # the receiver restamps from the credential (ch6-5) observation=args.observation, implication=args.implication, project=args.project or INAPPLICABLE, location=location, excerpt=body, story=args.story or INAPPLICABLE, depth=args.depth or INAPPLICABLE, condition_code=args.code_id or INAPPLICABLE, tags=tuple(args.tag or ()), ) except InvalidRecord as rejected: # ch6-4: reject at write, naming everything missing, while the writer still has context. print(f"rejected — {rejected}", file=sys.stderr) return 1 store = _store(args) path = store.file(record, queue=True) outcome = send_one(record, store, args.enrolment) print(f"filed {record.id}\n {path}") if outcome.sent: print(f" sent — {outcome.reason}") elif outcome.refused: # Never "sent". The record stays in the outbox; removing it is an operation (ch6-9). print(f" REFUSED — {outcome.reason}", file=sys.stderr) elif outcome.queued: print(f" queued — {outcome.reason}") else: store.dequeue(record.id) # not enrolled: local by default, nothing pending (ch6-6) print(" local only — this installation is not enrolled") return 0 def cmd_list(args) -> int: store = _store(args) queued = set(store.queued()) records = store.all() if not records: print(f"no records in {store.root}") return 0 for record in records: mark = "*" if record.id in queued else " " print(f"{mark} {record.id}") print(f" {record.observation[:100]}") print(f"\n{len(records)} record(s); {len(queued)} awaiting transmission (*)") return 0 def cmd_show(args) -> int: try: print(_store(args).get(args.id).to_json(), end="") except KeyError: print(f"no record {args.id!r}", file=sys.stderr) return 1 return 0 def cmd_withdraw(args) -> int: """``ch6-9`` — consent that cannot be taken back was never consent.""" if _store(args).withdraw(args.id): print(f"withdrawn {args.id}") else: print(f"no record {args.id!r} here (already withdrawn?)") return 0 def cmd_send(args) -> int: outcome = drain(_store(args), args.enrolment) if not outcome.sent and not outcome.queued: print(outcome.reason or "nothing queued") return 0 print(f"sent {len(outcome.sent)}, still queued {len(outcome.queued)}" + (f", REFUSED {len(outcome.refused)}" if outcome.refused else "")) if outcome.reason: print(f" {outcome.reason}") return 0 def cmd_why(args) -> int: print(instruction()) return 0 def cmd_init(args) -> int: """``ch9-5``, ``ch9-7`` — write what is missing, keep what is there, say what to do next.""" from plumb.scaffold import init, next_steps from plumb.spec_scaffold import ScaffoldUnavailable try: outcome = init(args.project, spec=args.spec) except ScaffoldUnavailable as unavailable: print(f"{unavailable}", file=sys.stderr) return 1 for name in outcome.written: print(f"wrote {name}") for name in outcome.kept: # Never overwritten, and never silent about it: a file the tool wrote once belongs to the # project the moment it lands, and the adopter has to know which half they are looking at. print(f"kept {name} (already present — left exactly as it is)") print(f"\ndetected {outcome.detected.adapter}") for note in outcome.detected.notes: print(f" still to do: {note}") print(next_steps(outcome.detected)) return 0 def cmd_registry(args) -> int: """``ch9-9`` — mint, list, revoke. Never a hand-edit of JSON on a live host.""" from plumb import registry_ops path = Path(args.registry) if args.command == "mint": try: credential = registry_ops.mint(path, args.label, args.relationship) except registry_ops.RegistryError as refused: print(f"{refused}", file=sys.stderr) return 1 # Shown once, here, and nowhere else — the registry keeps only a fingerprint of it. print(f"minted for {args.label!r} ({args.relationship})\n") print(f" {credential}\n") print("This is the only time it will be shown; the registry stores only its fingerprint.") print("Send it to the contributor, who enrols with it. Lost means mint another.") return 0 if args.command == "revoke": removed = registry_ops.revoke(path, args.label) print(f"revoked {args.label!r}" if removed else f"no contributor {args.label!r} enrolled") return 0 enrolled = registry_ops.entries(path) if not enrolled: print(f"nobody enrolled in {path}") return 0 for entry in enrolled: # Who, never what. print(f" {entry.label:<24} {entry.relationship}") print(f"\n{len(enrolled)} enrolled") return 0 def cmd_board(args) -> int: """Run the adapter, derive, emit. Every number printed is one the gate produced. ``--manifest`` skips the runner entirely and reads one that already exists. That is not a convenience: the core reads only the manifest (``ch2-1``), and a manifest this process did not produce is the only way to *demonstrate* that rather than assert it. It is also the seam every future adapter arrives through — a Java adapter emits this file and nothing else. """ import contextlib from plumb.core.board import Board, natural_key from plumb.core.gate import CONDITIONS from plumb.core.manifest import InvalidManifest, Manifest if args.manifest: try: manifest = Manifest.read(Path(args.manifest)) except (OSError, InvalidManifest, ValueError) as bad: print(f"--manifest: {bad}", file=sys.stderr) return 2 else: from plumb.config import Config, InvalidConfig from plumb.launch import LaunchFailed, produce try: config = Config.find(args.project) except InvalidConfig as bad: print(f"{bad}", file=sys.stderr) return 2 # The runner writes its own progress to stdout, which would corrupt the neutral format # on the way to a consumer (`ch5-2`). Its chatter goes to stderr; stdout carries only # the board. say = lambda line: print(line, file=sys.stderr, flush=True) try: with contextlib.redirect_stdout(sys.stderr): # No flag defers to what the project declared, which defaults to off (`ch4-2`). # The flag can only turn it *on*: a switch that silently overrode a project's own # declaration would make the board depend on how it was invoked. Progress goes to # stderr — a run measured in minutes that prints nothing looks like a hang. manifest = produce(config, pytest_args=args.pytest_args, mutation=args.mutation, report=say) except (LaunchFailed, InvalidConfig) as bad: print(f"{bad}", file=sys.stderr) return 2 if args.emit_manifest: Manifest.write(manifest, Path(args.emit_manifest)) print(f"manifest → {args.emit_manifest}", file=sys.stderr) args._adapter = config.adapter board = Board.render(manifest) if args.json: print(board.to_json()) return 0 if not board.verdicts: print(_nothing_cited(getattr(args, "_adapter", "python")), file=sys.stderr) print("\n══════ BOARD ══════") for sid, verdict in sorted(board.verdicts.items(), key=lambda kv: natural_key(kv[0])): depth = f" [{' · '.join(verdict.achieved)}]" if verdict.achieved else "" fact = board.manifest.stories[sid].mutation # A score, never a word: "survived" reads as a failing grade, and a story whose tests # check most of what they run has not failed anything. score = f" {fact.killed}/{fact.total} checked" if fact else "" print(f" {verdict.status.upper():28} {sid}{depth}{score}") for code, count in sorted(_conditions(board).items()): # Nothing is silently skipped: say what could not run, what to do, and what that looks # like (`ch3-7`, `ch0-7`). Each part is labelled so the reader can tell the diagnosis # from the instruction rather than parsing one long sentence. condition = CONDITIONS.get(code) print(f"\n {count} stories [{code}]") if condition: print(f" what · {condition.what}") print(f" next · {condition.next_step}") if condition.example: print(f" example · {condition.example}") for unlock in board.as_dict()["unlocks"]: # The board markets its own next step from its own output — no selling (`ch5-7`). print(f"\n unlock [{unlock['code']}]") print(f" what · {unlock['what']}") print(f" next · {unlock['next_step']}") print("\n " + " · ".join(f"{k}: {v}" for k, v in sorted(board.summary().items()))) print("═══════════════════\n") return 0 #: What to do when a board is empty, in the spelling of whichever adapter ran. The launcher may #: name a language; the core may not (`ch0-4`), which is why this lives here and not in the gate. _FIRST_CITATION = { "java": ' @Proves(value = "APP-1", depth = "unit")\n void yourTest() { ... }', "python": ' @pytest.mark.proves("APP-1", depth="unit")\n def test_your_thing(): ...', } def _nothing_cited(adapter: str) -> str: """An empty board is the most likely first run, and silence reads as a broken tool rather than as "you have not cited anything yet".""" return ( "\nNo stories are cited yet, so there is nothing to derive a board from.\n\n" "Cite one from a test — the id is yours, and Plumb keeps no catalog of them:\n\n" f"{_FIRST_CITATION.get(adapter, _FIRST_CITATION['python'])}\n\n" "Then say where your production code lives, in plumb.toml beside your project:\n\n" ' production = ["src/myapp"]\n' ' entry_points = ["myapp.cli:main"]\n\n' "Without entry_points a story can still pass; it just cannot be checked for reaching\n" "the wired system, and the board says so rather than pretending otherwise.\n" ) def _conditions(board) -> dict[str, int]: counts: dict[str, int] = {} for verdict in board.verdicts.values(): for code in verdict.conditions: counts[code] = counts.get(code, 0) + 1 return counts def build_parser() -> argparse.ArgumentParser: """``plumb `` — the board and friction capture. The nesting was added before it was needed, so that gaining a second group moved no existing command.""" parser = argparse.ArgumentParser(prog="plumb", description="Plumb — proof that a story is done") groups = parser.add_subparsers(dest="group", required=True) board = groups.add_parser("board", help="derive and emit the board from a test run") board.add_argument("--json", action="store_true", help="emit machine-readable JSON instead of the text board") board.add_argument("--manifest", metavar="FILE", help="derive from a manifest file instead of running the tests — how a " "run from another language gets a board") board.add_argument("--mutation", action="store_true", help="also check that the tests citing a story would notice its code " "being broken. Slow — minutes on a small project, longer on a real " "one; it prints the cost before starting. Never fails a build") board.add_argument("--emit-manifest", metavar="FILE", help="write the manifest the runner produced, to inspect or replay") board.add_argument("--project", metavar="DIR", default=".", help="the project to read plumb.toml from (default: this directory)") board.add_argument("pytest_args", nargs="*", help="passed to the runner, e.g. a path to limit collection. Put `--` " "first to pass the runner its own flags: plumb board -- -k orders") board.set_defaults(func=cmd_board) init = groups.add_parser("init", help="set a project up to run a board") init.add_argument("--spec", action="store_true", help="also emit a runnable spec — the claim vocabulary, a first chapter " "with its process flow, and the integrity gates that keep it honest") init.add_argument("--project", metavar="DIR", default=".", help="the project to write plumb.toml into (default: this directory)") init.set_defaults(func=cmd_init) registry = groups.add_parser( "registry", help="issue, list and revoke contributor credentials", description="Admitting a contributor is an operation, not a hand-edit of a file on a " "live host. A credential is shown exactly once and stored only as a " "fingerprint; the listing names who is enrolled and never what they hold.") registry.add_argument("--registry", default="registry.json", help="the registry file") reg = registry.add_subparsers(dest="command", required=True) minted = reg.add_parser("mint", help="issue a credential, shown once") minted.add_argument("label", help="who this is — what revocation names later") minted.add_argument("--relationship", default="trusted-tester") reg.add_parser("list", help="who is enrolled (never what they hold)") revoked = reg.add_parser("revoke", help="remove a contributor") revoked.add_argument("label") registry.set_defaults(func=cmd_registry) friction = groups.add_parser("friction", help="capture what rubbed while using the method", description=FRICTION_DESCRIPTION) friction.add_argument("--store", help=f"record directory (default: {DEFAULT_STORE})") friction.add_argument("--enrolment", help="enrolment file (default: ~/.config/plumb/enrolment.json)") sub = friction.add_subparsers(dest="command", required=True) f = sub.add_parser("file", help="file a friction record") f.add_argument("--observation", required=True, help="what rubbed, in your words") f.add_argument("--implication", required=True, help="what it means for the method — not just what happened") f.add_argument("--story", help="the story it bears on, in your own ids — e.g. APP-142") f.add_argument("--depth", help="unit | component | wiring | standards-integration | smoke") f.add_argument("--project", help="project name (default: none recorded)") f.add_argument("--code", metavar="PATH:START-END", help="carry the code that rubbed") f.add_argument("--root", help="path root, so the location reads relative") f.add_argument("--code-id", metavar="CODE", help="the condition code, if one was surfaced") f.add_argument("--tag", action="append", help="repeatable") f.add_argument("--id", help="override the minted id") f.set_defaults(func=cmd_file) for name, fn, help_text in ( ("list", cmd_list, "list records; * marks awaiting transmission"), ("send", cmd_send, "retry anything queued"), ("why", cmd_why, "print when to file"), ): p = sub.add_parser(name, help=help_text) p.set_defaults(func=fn) for name, fn, help_text in (("show", cmd_show, "print one record"), ("withdraw", cmd_withdraw, "remove a record")): p = sub.add_parser(name, help=help_text) p.add_argument("id") p.set_defaults(func=fn) return parser def main(argv=None) -> int: args = build_parser().parse_args(argv) return args.func(args) if __name__ == "__main__": raise SystemExit(main()) PK \f!plumb/config.py"""What a project declares about itself, in one file, read once by the core. ``plumb.toml`` is the single home. The **core owns configuration** and every adapter receives resolved values, which is what makes one command work across languages: the launcher decides what to run and how, and an adapter never learns a config format. That direction is deliberate — the Java adapter hand-writes its JSON rather than take a dependency, and asking it to parse TOML as well would buy a second format nobody needs. This module sits **outside** ``plumb.core`` on purpose. It knows adapter names, and the core may not (``ch0-4``, ``ch1-5``): the gate reads a manifest and nothing else, so a language named here must never be a language named there. Launcher knowledge belongs beside the launcher. The values are deliberately **adapter-native** (``ch3-U1``). ``App#handleRequest`` and ``myapp.cli:main`` name the same idea in two languages, and flattening them into one invented spelling would mean every adapter translating out of a form none of them use. """ import tomllib from dataclasses import dataclass, field from pathlib import Path PYTHON, JAVA = "python", "java" ADAPTERS = (PYTHON, JAVA) CONFIG_NAME = "plumb.toml" class InvalidConfig(ValueError): """Refused where it is read, naming the file and what it should say. The reader of these is somebody adopting the tool, so a refusal costs them one sentence and a guess costs them a board that is quietly wrong.""" @dataclass(frozen=True) class Config: """One project's declaration. Absent everywhere means an empty one, never an error — the lowest input level still produces a real board (``ch0-7``).""" adapter: str = PYTHON #: Where the wired system begins, in the adapter's own spelling (``ch3-2``). entry_points: tuple[str, ...] = () #: What counts as production code, in the adapter's own spelling. production: tuple[str, ...] = () #: Where that code's **source** lives. Distinct from ``production`` because the two are not #: the same question in every language: Python names directories and can mutate what it #: traced, while Java names classes and mutation needs the ``.java`` files behind them. sources: tuple[str, ...] = () #: ``ch4-2`` — off unless asked for. mutation: bool = False #: ``story id -> archetype`` (``ch1-7``, ``ch2-7``), for a project whose spec is not executed #: by the same run that produces the manifest — which today is every non-Python project. #: #: This is **the open half of the archetype question**, not its answer. The archetype is the #: spec's to declare, and a Python project declares it by running the spec; naming it here as #: well is a second home for one fact, and second homes drift. It is written down as config #: rather than left in an adapter-side file for one reason only: config has a single reader, #: so at least the drift is between two things the core can see. Whether the spec's #: declarations should be readable as *data* by any adapter is the decision this defers. archetypes: dict = field(default_factory=dict) #: The adapter's own table, passed through untouched. Nothing here is interpreted, so a new #: adapter needs no change to this module. options: dict = field(default_factory=dict) #: The directory the config was found in; every relative path is relative to it, not to the #: caller's working directory. root: Path = field(default_factory=Path) @classmethod def find(cls, start: Path | str = ".") -> "Config": """Read ``plumb.toml`` from ``start``, or return an empty config. No upward search. A tool that walks parent directories will eventually find *somebody's* config, and a board derived from a file the user did not know they had is worse than one that says nothing was declared. """ start = Path(start) path = start / CONFIG_NAME if not path.is_file(): _refuse_stale_pyproject(start) return cls(root=start) return cls.parse(path.read_text(), root=start) @classmethod def parse(cls, text: str, root: Path | str = ".") -> "Config": try: table = tomllib.loads(text) except tomllib.TOMLDecodeError as bad: raise InvalidConfig(f"{CONFIG_NAME} is not valid TOML — {bad}") from bad adapter = table.get("adapter", PYTHON) if adapter not in ADAPTERS: raise InvalidConfig( f"{CONFIG_NAME}: adapter {adapter!r} is not one of {list(ADAPTERS)}" ) production = _strings(table, "production") return cls( adapter=adapter, entry_points=_strings(table, "entry_points"), production=production, # Python names directories for both, so defaulting is right there and merely # convenient elsewhere — an adapter that needs them apart says so. sources=_strings(table, "sources") or production, mutation=bool(table.get("mutation", False)), archetypes=dict(table.get("archetypes", {})), options=table.get(adapter, {}), root=Path(root), ) def resolved(self, *paths: str) -> tuple[Path, ...]: """Project-relative paths made absolute against the config's own directory.""" return tuple((self.root / p).resolve() for p in paths) @property def marker(self) -> str: """What this project calls a citation. A convention, not a mechanism (``ch9-4``) — a project citing 315 requirements under its own name should not have to rewrite them all before the tool will read any of them.""" return self.options.get("marker", "proves") @property def declares_anything(self) -> bool: return bool(self.entry_points or self.production) def _strings(table: dict, key: str) -> tuple[str, ...]: value = table.get(key, ()) if isinstance(value, str) or not all(isinstance(v, str) for v in value): raise InvalidConfig(f"{CONFIG_NAME}: {key} must be a list of strings, got {value!r}") return tuple(value) def _refuse_stale_pyproject(start: Path) -> None: """A project that declared itself the old way is told where its declaration moved. Silently ignoring ``[tool.plumb]`` would hand that project an empty config and a board reading `not-checked` for everything — technically honest, and indistinguishable from "you configured nothing". They configured something; it is in the wrong file, and only this knows that. """ pyproject = start / "pyproject.toml" if not pyproject.is_file(): return try: table = tomllib.loads(pyproject.read_text()) except tomllib.TOMLDecodeError: return if table.get("tool", {}).get("plumb"): raise InvalidConfig( f"{pyproject} still declares [tool.plumb], and configuration now lives in " f"{CONFIG_NAME} beside it — one file, because the core reads it for every language " "and a Java project has no pyproject.toml. Move the keys across and delete the table" ) PK{\ plumb/core/PK=\]ccplumb/core/__init__.py"""The language-blind core: the manifest, the gate, and the board. Nothing here may import an adapter. That is ``ch0-4`` and ``ch2-1`` expressed as a package boundary rather than as a convention — the core reads only the manifest, so a new language is a new adapter and nothing else. ``tests/test_core_is_policy_free.py`` asserts it by inspection. """ PK{\*#Fplumb/core/board.py"""Chapter 5's outbound edge: the board. The core's derived, tool-facing report — per story, its status, its achieved depth vector, and whatever the run could not check (``ch5-1``). **The manifest is internal; the board is the public contract**, so this is the thing other tools are allowed to depend on. Emitted in a documented, neutral JSON serialization (``ch5-2``) under an explicit ``board_version``, because external tools depend on the shape: it evolves compatibly and a breaking change is a new version, never a silent shift (``ch5-6``). The version is deliberately its own, separate from the manifest's — one is a public contract and the other is an internal one, and tying them would make an internal refactor look like a breaking change. Nothing here decides anything. It renders what the gate derived, so the number a consumer reads is the number the gate produced. """ import json import re from dataclasses import dataclass, field from plumb.core.gate import CONDITIONS, UNLOCKS, Verdict, derive_all, unlocks from plumb.core.manifest import Manifest BOARD_VERSION = 1 _DIGITS = re.compile(r"(\d+)") def natural_key(story_id: str) -> tuple: """Order ids the way a reader expects: ``ch6-2`` before ``ch6-10``. Presentation only. It never affects a status, and it **requires** no shape — digit runs compare as numbers and everything else as text, so ``PROJ-7`` and ``SHEET-42`` order sensibly too. That keeps ``ch2-2``'s promise intact: the id stays opaque and no catalog is needed. Lexical ordering is the alternative, and it puts ``ch6-10`` in the middle of the single digits, which reads as a bug in the board rather than a property of sorting. """ return tuple( (1, int(part)) if part.isdigit() else (0, part) for part in _DIGITS.split(story_id) if part ) def _mutation(fact) -> dict | None: """A score the consumer can put in proportion, never a word it has to interpret.""" if fact is None: return None return {"total": fact.total, "killed": fact.killed, "survivors": fact.survivors} def _signal(code: str, registry: dict, n: int | None = None) -> dict: """Emit a code's tagged parts. An unknown code still emits, carrying only itself — a consumer keying off the code is not broken by prose it has never seen (``ch5-8``).""" known = registry.get(code) fill = (lambda s: s.format(n=n)) if n is not None else (lambda s: s) return { "code": code, "what": fill(known.what) if known else "", "next_step": known.next_step if known else "", "example": known.example if known else "", } @dataclass(frozen=True) class Board: """The rendered result, ready to emit (``ch5-1``).""" verdicts: dict[str, Verdict] = field(default_factory=dict) manifest: Manifest = field(default_factory=Manifest) board_version: int = BOARD_VERSION @classmethod def render(cls, manifest: Manifest) -> "Board": return cls(verdicts=derive_all(manifest), manifest=manifest) def as_dict(self) -> dict: observer = self.manifest.observer return { "board_version": self.board_version, # Which instrument made these measurements. Two boards are only comparable if the # same kind of observer produced them (`ch2-11`). "observer": None if observer is None else { "adapter": observer.adapter, "version": observer.version, "runtime": observer.runtime, "mechanism": observer.mechanism, }, "stories": { sid: { "status": v.status, "achieved": list(v.achieved), # Each surfaced condition carries its stable code beside its prose, and # the prose is split into what happened, what to do, and what that looks # like. The code is the promise; the rest may be reworded (`ch5-8`). "conditions": [_signal(c, CONDITIONS) for c in v.conditions], # `null` where the opt-in was off, which is *not asked* rather than a pass — # a capability whose result the board never shows is one a reader has to go # and find in the manifest (`ch4-5`, `ch2-10`). "mutation": _mutation(self.manifest.stories[sid].mutation), # Evidence, not opinion: what was proven and *how*, so a reviewer can act # on it rather than take the verdict on trust (`ch5-3`). "evidence": [ { "test": c.test, "depth": c.depth, "result": c.result, "grounding": c.grounding, "ref": c.ref, } for c in self.manifest.stories[sid].citations ], } for sid, v in sorted(self.verdicts.items(), key=lambda kv: natural_key(kv[0])) }, # What this run has earned the right to suggest next, derived from the run itself # rather than advertised (`ch5-7`). "unlocks": [ _signal(code, UNLOCKS, n) for code, n in sorted(unlocks(self.verdicts, self.manifest).items()) ], } def to_json(self, indent: int | None = 2) -> str: # Not ``sort_keys``: that would re-impose the lexical order this deliberately avoids. # Insertion order is already deterministic, so emitted boards still diff cleanly. return json.dumps(self.as_dict(), indent=indent) def summary(self) -> dict[str, int]: """Counts per status — an aggregate that asserts nothing the source does not.""" counts: dict[str, int] = {} for v in self.verdicts.values(): counts[v.status] = counts.get(v.status, 0) + 1 return counts PK\(\w0w0plumb/core/gate.py"""Chapter 1's core: the proof gate, and the status derived from it. **Status is derived, never authored** (``ch1-4``) — computed live from the manifest on every run and never stored, so a proof that later fails un-proves its story and no stale "done" can exist. Nothing here writes a status anywhere; the caller renders what it returns. The gate is two checks in series, fail-fast (``ch1-2``): a citing test **ran and passed** (``ch1-2-1``), and its execution matched what the story's archetype claims — behavioral must **ground**, structural must be **inert**. The second check is never *waived*, only *mirrored*: an archetype that switched grounding off would be the exemption every fake wants. This module imports the manifest and nothing else. That is the whole of ``ch2-1`` and ``ch0-4`` in one line — the core reads only the manifest, so it touches no test framework and knows no language. """ from dataclasses import dataclass from plumb.core.manifest import ( BEHAVIORAL, DISPATCHED, GROUNDED, UNGROUNDED, INERT, NOT_CHECKED, Manifest, Story, ) # The derived status vocabulary (`ch1-4`). PROVEN = "proven" UNPROVEN = "unproven" PASSED_WIRING_NOT_VERIFIED = "passed-wiring-not-verified" # Surfaced conditions carry a stable code beside their prose, because a consumer keys off the # code and never off a sentence that will be reworded. `ch5-8` generalizes this; `ch3-7` is the # one condition the gate can already raise. NO_ENTRY_POINTS = "grounding.not-checked" DEPTH_UNSUPPORTED = "depth.unsupported" REACHED_OUTSIDE = "grounding.reached-outside" #: Rungs whose claim is *about* reaching the wired system. A citation claiming one of these whose #: execution never grounded is lying about its depth, and grounding is the mechanical check on an #: otherwise-trusted claim (``ch3-6``). #: #: This lives in the core rather than beside the tracer, and that placement is the whole reason it #: works: it is a pure function of two manifest fields, but while it sat in the adapter the gate #: could not import it (``ch1-5`` forbids core reaching into an adapter) — so the only caller that #: could ever have used it was prevented by the seam, and two passing tests made it look alive. WIRED_RUNGS = frozenset({"wiring", "standards-integration", "smoke", "smoke/e2e", "e2e"}) def contradicts_claimed_depth(depth: str, grounding: str) -> bool: """``ch3-6`` — is this citation's claimed rung refuted by what actually executed? Only the wired rungs make a claim grounding can check. A ``unit`` or ``component`` claim is about isolation and is *expected* not to ground, so it is never contradicted here — the remainder stays trusted and is named as the residue it is (``ch1-2-U2``). """ if depth not in WIRED_RUNGS: return False return grounding in (UNGROUNDED, INERT) WORK_DISPATCHED = "grounding.dispatched" @dataclass(frozen=True) class Condition: """A surfaced gap, in three separately-addressable parts. They are kept apart rather than run into one sentence because they answer different questions and are consumed differently: a reader wants ``what``, someone fixing it wants ``next_step``, and someone who has never seen the syntax needs ``example``. A consumer can render or suppress each on its own, and only ``code`` is a promise — the prose may be reworded freely (``ch5-8``). """ code: str what: str next_step: str example: str = "" UNLOCK_GROUNDING = "unlock.grounding" UNLOCK_MUTATION = "unlock.mutation" #: Every condition the gate can raise. The set is finite and knowable, so what the board can #: say is a list rather than an open question (`ch5-8`). CONDITIONS = { WORK_DISPATCHED: Condition( code=WORK_DISPATCHED, what="the work was handed to a thread this adapter could not follow, so whether it ran " "under an entry point is unknown", next_step="carry the entry-point context across that hand-off, or accept that this story " "is proven no deeper than its own thread", example="an executor created before the entry point ran inherits nothing — wrapping it " "is what carries the context across", ), REACHED_OUTSIDE: Condition( code=REACHED_OUTSIDE, what="this story grounded, and the same test also ran production code that no entry " "point reaches — so the grounding does not vouch for all of it", next_step="check that what this story claims is the part that ran wired, not the part " "that did not; splitting the test is the usual answer", example="a test that drives an entry point and then calls a helper directly grounds on " "the first call, and nothing here can tell the two apart", ), DEPTH_UNSUPPORTED: Condition( code=DEPTH_UNSUPPORTED, what="a citation claims it reached the wired system, and its execution never did", next_step="lower the claimed depth to what the test actually reaches, or drive the story " "through a declared entry point", # Language-neutral on purpose. Every one of these is printed to whoever ran the command, # and the citation syntax differs per language — a Java user was being shown a pytest # marker, which is advice they cannot act on and cannot tell is not meant for them. example='cite it at depth="unit" instead — a claim the execution supports', ), NO_ENTRY_POINTS: Condition( code=NO_ENTRY_POINTS, what="grounding did not run — no production entry points are declared", next_step="declare where the wired system begins, to catch orphaned code", # A form without a home is advice nobody can act on, so the example names the file. The # *values* stay in the adapter's own spelling (`ch3-U1`) and the file does not, which is # what lets one hint serve every language. example='in plumb.toml: entry_points = ["myapp.cli:main"]', ), } #: What more this project could get, and what it would cost them to get it. The board markets #: its own next step from its own output, so nobody has to be sold anything (`ch5-7`). UNLOCKS = { UNLOCK_GROUNDING: Condition( code=UNLOCK_GROUNDING, what="{n} stories passed but their wiring was never checked", next_step="declare production entry points, and those stories get gated on real wiring", example='in plumb.toml: entry_points = ["myapp.cli:main"]', ), UNLOCK_MUTATION: Condition( code=UNLOCK_MUTATION, what="{n} stories are proven, and nothing has checked that their tests assert anything", next_step="run mutation over them to find a test that runs the code without checking it", example="plumb board --mutation — off by default because it is slow, not because it is " "optional; it prints what it will cost before it starts", ), } #: Everything the board is permitted to say. A consumer branches on these and never on prose. ALL_CODES = frozenset(CONDITIONS) | frozenset(UNLOCKS) def unlocks(verdicts: dict[str, "Verdict"], manifest: Manifest) -> dict[str, int]: """Which next steps this run has earned the right to suggest, and how many stories each would move. Derived from the run — never a standing advertisement (`ch5-7`). The manifest is read as well as the verdicts because a suggestion must not outlive its own answer: a story whose citations already carry a mutation fact has had the question asked, and telling that project to turn mutation on is the board asserting a gap its own source does not show (`ch7-8`). """ found: dict[str, int] = {} partial = sum(1 for v in verdicts.values() if v.status == PASSED_WIRING_NOT_VERIFIED) if partial: found[UNLOCK_GROUNDING] = partial unchecked = sum(1 for sid, v in verdicts.items() if v.status == PROVEN and not _mutation_checked(manifest.stories[sid])) if unchecked: found[UNLOCK_MUTATION] = unchecked return found def _mutation_checked(story: Story) -> bool: """Has the question been asked of this story at all? An unfilled slot is *not asked* (`ch2-10`), which is exactly when the next step is still available.""" return story.mutation is not None @dataclass(frozen=True) class Verdict: """What the gate derived for one story — the output, computed each run (``ch1-4``).""" story: str status: str achieved: tuple[str, ...] = () conditions: tuple[str, ...] = () @property def proven(self) -> bool: return self.status == PROVEN def _second_check(archetype: str, grounding: str) -> str: """The archetype selects a check over the *same* per-test fact (``ch1-2-4``). Every arm of both branches is handled deliberately. An unhandled arm is exactly where a case falls silently through a gate, and this framework already shipped that bug once — its structural branch had no arm for one of the grounding facts (``ch7-9``). """ if archetype == BEHAVIORAL: if grounding == GROUNDED: return PROVEN if grounding in (NOT_CHECKED, DISPATCHED): # Two different reasons for the same verdict. `not-checked`: nothing declared where # the real system begins. `dispatched`: it was declared, the check ran, and the work # left for a thread the adapter could not follow. Neither is a failure and neither is # a pass — degrade and surface (`ch3-7`, `ch0-7`). They are told apart by the # condition the board carries, because the reader's next move differs. return PASSED_WIRING_NOT_VERIFIED # `ungrounded` is the orphaned-"done" fake; `inert` never ran the code at all. return UNPROVEN # Structural: proved by inspection, so the citing test must be inert (`ch1-2-3`). Anything # that executed production code was not inspecting the shape — including `not-checked`, # which means production code *did* run, merely without a declared surface to judge it by. return PROVEN if grounding == INERT else UNPROVEN def derive(story_id: str, story: Story) -> Verdict: """Run the gate over one story's citations and return its derived status.""" passing = [c for c in story.citations if c.ran_and_passed] if not passing: # No citing test at all, or every one skipped or failed — a claim, not a proof. return Verdict(story_id, UNPROVEN) outcomes = {_second_check(story.archetype, c.grounding) for c in passing} status = ( PROVEN if PROVEN in outcomes else PASSED_WIRING_NOT_VERIFIED if PASSED_WIRING_NOT_VERIFIED in outcomes else UNPROVEN ) # The achieved vector is the rungs a story is *actually* proven at (`ch1-3`) — derived and # shown, so unit-only reads as unit-only. achieved = tuple(sorted({c.depth for c in passing if c.depth})) conditions = () # ch3-6: a claimed rung the execution refutes is surfaced whatever the status — a story can be # proven by one citation while another lies about how deep it reached. if any(contradicts_claimed_depth(c.depth, c.grounding) for c in passing): conditions += (DEPTH_UNSUPPORTED,) # ch3-5's stated limit, made visible rather than left to be discovered: grounded on the # strength of one call, while another went somewhere no entry point reaches. if any(c.grounding == GROUNDED and c.reached_outside_entry for c in passing): conditions += (REACHED_OUTSIDE,) if status == PASSED_WIRING_NOT_VERIFIED: # Name the reason, not just the verdict: "declare entry points" is useless advice to a # project that already did, and told only that, they would go looking in the wrong place. reasons = {c.grounding for c in passing} conditions += tuple(sorted( ({WORK_DISPATCHED} if DISPATCHED in reasons else set()) | ({NO_ENTRY_POINTS} if NOT_CHECKED in reasons else set()) )) return Verdict(story_id, status, achieved, conditions) def derive_all(manifest: Manifest) -> dict[str, Verdict]: """Derive every story's status from the manifest — the core's whole job.""" return {sid: derive(sid, story) for sid, story in manifest.stories.items()} PK{\R;;plumb/core/manifest.py"""Chapter 2's hub: the language-neutral contract between adapter and core. The **one** thing the core reads (``ch2-1``). An adapter produces it from a test run; the gate reasons over it. Because it is the sole interface, the core touches no runner and knows no language — a new language is a new adapter and nothing else (``ch0-4``). It carries **facts, not verdicts** (``ch2-5``): the adapter records what happened, and only the gate decides proven from unproven. Nothing here computes a status, and no adapter can stamp one. Grounding and mutation are **slots** (``ch2-6``) — this module fixes where they sit, never how they are produced (chapters 3 and 4). Two fields are not facts about the run at all: a story's ``archetype`` is read off the spec (``ch2-7``), and a citation's ``ref`` is the consumer's own external key, which Plumb carries and never parses (``ch2-8``). """ import json from dataclasses import asdict, dataclass, field from pathlib import Path # The runner's own signal, normalized — the core invents no status model (`ch2-4`). PASSED, SKIPPED, FAILED = "passed", "skipped", "failed" RESULTS = (PASSED, SKIPPED, FAILED) # The grounding slot's vocabulary (`ch3-4`). `not-checked` is the honest answer when no entry # points are declared — surfaced, never silently skipped (`ch3-7`). GROUNDED, UNGROUNDED, INERT, NOT_CHECKED = "grounded", "ungrounded", "inert", "not-checked" #: The work left for a thread the adapter could not follow. Distinct from `not-checked` — that is #: "the question could not be asked", this is "it was asked and the answer walked away" — and the #: two need different things from the reader (`ch3-4`). DISPATCHED = "dispatched" GROUNDING = (GROUNDED, UNGROUNDED, INERT, NOT_CHECKED, DISPATCHED) # What kind of claim a story makes (`ch1-7`). Behavioral is the default, so a spec that never # mentions archetypes is unchanged. BEHAVIORAL, STRUCTURAL = "behavioral", "structural" ARCHETYPES = (BEHAVIORAL, STRUCTURAL) SCHEMA_VERSION = 4 # Each schema version's shape, stated once and in full (`ch2-9`). A version's entry is the whole # truth about that version: adding a field means adding a **new entry**, never editing an old one, # because two cores that both claim to read version 1 must agree on what version 1 is. # # Checking against the *declared* version rather than against whatever the current dataclass # happens to define is what makes a refusal accurate. The looser check — "no version I know # defines this field" — reads identically while only one version exists and starts lying the day # a second one lands. MANIFEST_FIELDS = { 1: frozenset({"schema_version", "stories"}), 2: frozenset({"schema_version", "stories", "observer"}), 3: frozenset({"schema_version", "stories", "observer"}), 4: frozenset({"schema_version", "stories", "observer"}), } _V13_STORY = frozenset({"archetype", "citations"}) STORY_FIELDS = {1: _V13_STORY, 2: _V13_STORY, 3: _V13_STORY, # Version 4 moves mutation from the citation to the story, where `ch4-5` always # said it belonged: a mutant is run against *every* test citing the story, so the # result is one fact about the story and repeating it per citation would be the # same number written down N times. 4: _V13_STORY | {"mutation"}} _V2_CITATION = frozenset({"test", "depth", "result", "grounding", "mutation", "ref"}) CITATION_FIELDS = {1: _V2_CITATION, 2: _V2_CITATION, 3: _V2_CITATION | {"reached_outside_entry"}, 4: (_V2_CITATION - {"mutation"}) | {"reached_outside_entry"}} #: A citation cannot be reconstructed without these. The rest carry defaults, so an adapter that #: knows nothing about grounding or mutation still emits a valid document (`ch0-7`). REQUIRED_CITATION_FIELDS = {1: ("test", "depth", "result"), 2: ("test", "depth", "result"), 3: ("test", "depth", "result"), 4: ("test", "depth", "result")} #: Derived, so a version cannot be described and left unsupported, or supported and left #: undescribed. The asymmetry is the point: a manifest **older** than the core is readable — the #: core still knows that shape — while one **newer** is not, since nothing can be inferred from a #: description that has not been written yet. SUPPORTED_VERSIONS = frozenset(CITATION_FIELDS) class InvalidManifest(ValueError): """The manifest is the sole interface, so a malformed one is caught at its edge rather than misread downstream by a core that cannot see where it came from.""" @dataclass(frozen=True) class Provenance: """The instrument that made a manifest's measurements (``ch2-11``). ``mechanism`` is the load-bearing field, not ``runtime``. One runtime can offer two ways to watch the same thing whose visibility does not agree — CPython's ``sys.settrace`` sees one thread and ``settrace_all_threads`` sees them all — and a grounding fact says nothing about which was used. Two facts are comparable when the same kind of observer produced them, so this is what makes them comparable at all. """ adapter: str version: str runtime: str mechanism: str @dataclass(frozen=True) class Mutation: """How many mutants of a story's grounded code its tests noticed (``ch4-5``). **A count, not a verdict**, and that is the whole of what a boolean got wrong. A story's tests typically run far more code than they check — a component test drives an entry point and asserts one thing about the result — so "did any mutant survive?" is ``yes`` for almost every real test, and a fact that is almost always the same value carries almost no information. A score separates the test that checks nothing from the one that checks most of what it runs, and those need different things from the reader. It stays advisory whatever the numbers say (``ch4-6``): an *equivalent* mutant is indistinguishable from a real survivor (``ch4-U2``), so a shortfall is a place to look and never a verdict to act on unread. """ total: int killed: int def __post_init__(self): if self.total < 1: raise InvalidManifest(f"mutation total {self.total} — a run that made no mutant " "reports nothing at all, never a score out of zero") if not 0 <= self.killed <= self.total: raise InvalidManifest(f"mutation killed {self.killed} is not within 0..{self.total}") @property def survivors(self) -> int: return self.total - self.killed @dataclass(frozen=True) class Citation: """One citing test's facts for a story (``ch2-3``). ``test`` is opaque and adapter-native — a pytest node id, a JUnit method handle. The core carries it and never parses it, which is also why it cannot serve as anyone's join key: it is derived from a path and a name, so it moves when a test is renamed. ``ref`` exists for that (``ch2-8``). """ test: str depth: str result: str grounding: str = NOT_CHECKED ref: str | None = None #: Did this test also run production code that no entry point reached? Grounding is a #: property of the whole test (``ch3-1``), so a test can ground on one call and exercise #: unreachable code in the next; this is what stops that passing silently (``ch3-5``). #: ``None`` is *not reported* — an adapter that cannot observe it says nothing (``ch2-10``). reached_outside_entry: bool | None = None def __post_init__(self): if self.result not in RESULTS: raise InvalidManifest(f"{self.test}: result {self.result!r} is not one of {RESULTS}") if self.grounding not in GROUNDING: raise InvalidManifest(f"{self.test}: grounding {self.grounding!r} is not one of {GROUNDING}") @property def ran_and_passed(self) -> bool: """``ch1-2-1`` — a skip or an abort is a claim, not proof.""" return self.result == PASSED @dataclass(frozen=True) class Story: """A story's entry: its declared archetype, and every test that cited it. Keyed in the manifest by an **opaque** id (``ch2-2``) — a ``chN-M``, a ticket number, a spreadsheet row. The core never parses it and needs no catalog to build this. """ archetype: str = BEHAVIORAL citations: tuple[Citation, ...] = field(default_factory=tuple) #: ``None`` is *not asked* (``ch2-10``) — mutation is opt-in and off by default (``ch4-2``), #: and an unfilled slot must never read as a story whose tests checked nothing. mutation: Mutation | None = None def __post_init__(self): if self.archetype not in ARCHETYPES: raise InvalidManifest(f"archetype {self.archetype!r} is not one of {ARCHETYPES}") @dataclass(frozen=True) class Manifest: """``story id -> {archetype, [citations]}`` — the sole thing the core reads (``ch2-1``).""" stories: dict[str, Story] = field(default_factory=dict) #: Absent is a legal answer and means *not reported* — never inferred from whatever the #: reader happens to be running (`ch2-11`, `ch2-10`). observer: Provenance | None = None schema_version: int = SCHEMA_VERSION def to_json(self, indent: int | None = 2) -> str: return json.dumps(asdict(self), indent=indent, sort_keys=True) @classmethod def from_json(cls, text: str) -> "Manifest": """Read a manifest somebody else may have written (``ch2-9``). Every rejection names what is supported, because the reader of these messages is an adapter author with no view of this source. A refusal costs them one clear sentence; guessing would cost them a board that is subtly wrong and silent about why. """ raw = json.loads(text) version = raw.get("schema_version") _check_version(version) _reject_unknown("manifest", raw, MANIFEST_FIELDS[version], version) return cls( stories={ sid: _story(sid, entry, version) for sid, entry in raw.get("stories", {}).items() }, # A version 1 manifest could not express this, so carrying it forward leaves it # absent — the one value a migration may supply, meaning "not reported" (`ch2-10`). observer=_provenance(raw.get("observer")), ) def write(self, path: Path) -> Path: path.write_text(self.to_json()) return path @classmethod def read(cls, path: Path) -> "Manifest": return cls.from_json(Path(path).read_text()) def _check_version(version: object) -> None: if version in SUPPORTED_VERSIONS: return supported = ", ".join(str(v) for v in sorted(SUPPORTED_VERSIONS)) if version is None: raise InvalidManifest( f"no schema_version — every manifest declares the shape it was written against " f"(this Plumb reads: {supported})" ) if isinstance(version, int) and version > max(SUPPORTED_VERSIONS): raise InvalidManifest( f"schema_version {version} is newer than this Plumb understands (reads: {supported}). " "Upgrade Plumb, or emit an older version — a newer shape cannot be guessed at" ) raise InvalidManifest(f"schema_version {version!r} is not one this Plumb reads ({supported})") def _reject_unknown(where: str, raw: dict, allowed: frozenset[str], version: int) -> None: """Within a declared version the field set is fixed, so an unrecognised key means the producer changed shape without changing the number it claims (``ch2-9``). Applied to **stories as well as citations**. Silently dropping an unknown story field is worse than dropping an unknown citation field, not better: a typo like ``archetypes`` would leave the story quietly behavioral, and a mis-declared archetype changes which check the gate applies and therefore the status. """ unknown = sorted(set(raw) - allowed) if unknown: raise InvalidManifest( f"{where}: field(s) {unknown} are not defined by schema_version {version} " f"— it defines {sorted(allowed)}" ) def _story(story_id: str, raw: object, version: int) -> Story: if not isinstance(raw, dict): raise InvalidManifest(f"{story_id}: a story must be an object, got {type(raw).__name__}") _reject_unknown(story_id, raw, STORY_FIELDS[version], version) return Story( archetype=raw.get("archetype", BEHAVIORAL), citations=tuple(_citation(story_id, c, version) for c in raw.get("citations", ())), mutation=_mutation(story_id, raw.get("mutation")), ) def _mutation(story_id: str, raw: object) -> Mutation | None: if raw is None: return None if not isinstance(raw, dict): raise InvalidManifest(f"{story_id}: mutation must be an object with 'total' and 'killed', " f"got {type(raw).__name__}") missing = [f for f in ("total", "killed") if f not in raw] if missing: raise InvalidManifest(f"{story_id}: mutation is missing {missing} — a score needs both " "halves, and a survivor count nobody can put in proportion is noise") return Mutation(total=raw["total"], killed=raw["killed"]) def _citation(story_id: str, raw: object, version: int) -> Citation: if not isinstance(raw, dict): raise InvalidManifest(f"{story_id}: a citation must be an object, got {type(raw).__name__}") _reject_unknown(f"{story_id}: citation", raw, CITATION_FIELDS[version], version) missing = [f for f in REQUIRED_CITATION_FIELDS[version] if f not in raw] if missing: raise InvalidManifest(f"{story_id}: citation is missing required field(s) {missing}") fields = {k: v for k, v in raw.items() if k != "mutation"} # Versions 1-3 defined mutation on the citation as a bare outcome word. No adapter ever filled # it, so an empty slot carries forward exactly and a filled one is refused rather than guessed # at: "survived" says a mutant lived without saying how many ran, and inventing the # denominator would put a fabricated score in front of a reader who cannot see it was invented. if raw.get("mutation") is not None: raise InvalidManifest( f"{story_id}: citation carries mutation {raw['mutation']!r}, which schema_version " f"{version} could not put a count against. Re-run mutation and emit schema_version " f"{SCHEMA_VERSION}, where the score is the story's" ) return Citation(**fields) def _provenance(raw: object) -> Provenance | None: if raw is None: return None if not isinstance(raw, dict): raise InvalidManifest(f"observer must be an object, got {type(raw).__name__}") missing = [f for f in ("adapter", "version", "runtime", "mechanism") if f not in raw] if missing: raise InvalidManifest(f"observer is missing {missing} — a half-named instrument names none") return Provenance(**{k: str(raw[k]) for k in ("adapter", "version", "runtime", "mechanism")}) PK}\& & plumb/enrolment.py"""Who is filing, and whether their records may leave (``ch6-5``, ``ch6-6``). Consent is established **once, at enrolment**, never re-asked per record. An installation nobody enrolled keeps everything local and never transmits — the core ships to anyone, so that is the default case and it is what makes "nothing phones home" a guarantee rather than a promise. On ``ch6-5``'s "never from local config": the *client* cannot be the authority on who it is — anything on this disk is a claim by the filer. So the client states a relationship and the **receiver overwrites it** from the credential it authenticated. Local records are ``self`` because you are the only authority over your own store; transmitted ones are stamped by whoever accepted them. """ import json import os from dataclasses import dataclass from pathlib import Path RELATIONSHIPS = ("self", "trusted-tester") DEFAULT_ENROLMENT = Path( os.environ.get("PLUMB_ENROLMENT") or Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / "plumb" / "enrolment.json" ) class NotEnrolled(Exception): """No enrolment, so nothing transmits. Not an error — the default (``ch6-6``).""" @dataclass(frozen=True) class Enrolment: relationship: str credential: str endpoint: str def __post_init__(self): if self.relationship not in RELATIONSHIPS: raise ValueError(f"relationship must be one of {RELATIONSHIPS}, got {self.relationship!r}") if not self.credential.strip(): raise ValueError("an enrolment without a credential cannot carry consent") if not self.endpoint.startswith(("http://", "https://")): raise ValueError(f"endpoint must be an absolute http(s) URL, got {self.endpoint!r}") @property def may_transmit(self) -> bool: """Enrolment *is* the consent (``ch6-6``); there is no per-record question to ask.""" return True def to_json(self) -> str: return json.dumps( {"relationship": self.relationship, "credential": self.credential, "endpoint": self.endpoint}, indent=2, ) + "\n" def load(path: Path | None = None) -> Enrolment: """The enrolment, or ``NotEnrolled`` — which is the ordinary case, not a failure.""" path = Path(path or DEFAULT_ENROLMENT) if not path.is_file(): raise NotEnrolled(f"no enrolment at {path}; records stay local") raw = json.loads(path.read_text()) missing = {"relationship", "credential", "endpoint"} - set(raw) if missing: raise NotEnrolled(f"{path} is missing {sorted(missing)}") return Enrolment(raw["relationship"], raw["credential"], raw["endpoint"]) def is_enrolled(path: Path | None = None) -> bool: try: load(path) return True except (NotEnrolled, ValueError, json.JSONDecodeError): return False PKN}\ݕJJplumb/friction.py"""Chapter 6's record: friction, captured as data (``ch6-2``, ``ch6-3``, ``ch6-4``, ``ch6-11``). Knowledge, never evidence — nothing here touches the manifest or a story's status (``ch6-1``). A record is **validated at write** and rejected naming what is missing, because the writer is usually an agent still holding the context and that is the only moment a correction ever happens. Two field classes, per ``ch6-4``: * **required** — a record without these is not a record; * **conditional** — carried when they apply, and otherwise marked ``INAPPLICABLE`` (it did not apply) or ``UNCAPTURED`` (it applied and was lost). A blank, an inapplicable field and a lost one are three different facts; a schema that cannot tell them apart invites a writer to invent a plausible value, and an importer to invent one on a record's behalf. """ import json from dataclasses import asdict, dataclass, field, fields from datetime import datetime, timezone from pathlib import Path MAX_EXCERPT_LINES = 40 INAPPLICABLE = "n/a" # the field did not apply to this rub UNCAPTURED = "unknown" # it applied, and was never recorded — imports, mostly ABSENT = (INAPPLICABLE, UNCAPTURED) SOURCES = ("self", "trusted-tester") REQUIRED = ("id", "observed_at", "source", "observation", "implication") CONDITIONAL = ( "project", "commit", "location", "excerpt", "story", "depth", "condition_code", "board_status", ) class InvalidRecord(ValueError): """Raised at write, naming every missing field at once — a writer correcting one rejection at a time is a writer who stops filing.""" @dataclass(frozen=True) class FrictionRecord: id: str observed_at: str source: str observation: str implication: str project: str = INAPPLICABLE commit: str = INAPPLICABLE location: str = INAPPLICABLE excerpt: str = INAPPLICABLE story: str = INAPPLICABLE depth: str = INAPPLICABLE condition_code: str = INAPPLICABLE board_status: str = INAPPLICABLE tags: tuple[str, ...] = field(default_factory=tuple) def __post_init__(self): self.validate() def validate(self) -> None: problems = [*self._missing_required(), *self._malformed()] if problems: raise InvalidRecord("; ".join(problems)) def _missing_required(self) -> list[str]: return [ f"{name} is required and empty" for name in REQUIRED if not str(getattr(self, name)).strip() or getattr(self, name) in ABSENT ] def _malformed(self) -> list[str]: problems = [] if self.source not in SOURCES: problems.append(f"source must be one of {SOURCES}, got {self.source!r}") if not _is_utc(self.observed_at): problems.append(f"observed_at must be an ISO-8601 UTC instant, got {self.observed_at!r}") blank = [f.name for f in fields(self) if getattr(self, f.name) == ""] if blank: problems.append( f"blank is not a value — use {INAPPLICABLE!r} (did not apply) or " f"{UNCAPTURED!r} (applied, not recorded): {blank}" ) return problems @property def anchors(self) -> dict[str, str]: """The conditional fields that actually apply — what makes the rub resolvable later.""" return { name: getattr(self, name) for name in CONDITIONAL if getattr(self, name) not in ABSENT } def to_json(self) -> str: return json.dumps(asdict(self), indent=2, ensure_ascii=False) + "\n" @classmethod def from_json(cls, text: str) -> "FrictionRecord": raw = json.loads(text) raw["tags"] = tuple(raw.get("tags", ())) known = {f.name for f in fields(cls)} unknown = set(raw) - known if unknown: raise InvalidRecord(f"unknown field(s): {sorted(unknown)}") absent = [name for name in REQUIRED if name not in raw] if absent: raise InvalidRecord("; ".join(f"{name} is required and empty" for name in absent)) return cls(**raw) def write_to(self, directory: Path) -> Path: """One record, one file — which is what makes ``ch6-9`` removal an operation rather than an excavation.""" path = Path(directory) / f"{self.id}.json" path.write_text(self.to_json()) return path def _is_utc(stamp: str) -> bool: try: parsed = datetime.fromisoformat(stamp) except (TypeError, ValueError): return False return parsed.tzinfo is not None and parsed.utcoffset() == timezone.utc.utcoffset(None) def load_all(directory: Path) -> list[FrictionRecord]: return [FrictionRecord.from_json(p.read_text()) for p in sorted(Path(directory).glob("*.json"))] def excerpt(path: Path | str, start: int, end: int, *, root: Path | str | None = None) -> tuple[str, str]: """``(location, excerpt)`` for the lines that rubbed — read-only, and bounded. The record carries the code rather than a reference to it (``ch6-2``), so a reader a year later needs neither the repository nor permission to write to it (``ch6-12``). Lines are numbered, so the excerpt locates itself even quoted out of context. Bounded on purpose: an excerpt that needs the whole file is not pointing at the rub. That is the same bar ``ch6-3`` sets against volume, applied to code instead of prose. """ path = Path(path) if start < 1 or end < start: raise ValueError(f"not a line range: {start}-{end}") if end - start + 1 > MAX_EXCERPT_LINES: raise ValueError( f"{end - start + 1} lines exceeds the {MAX_EXCERPT_LINES}-line bar — an excerpt that " "large is not pointing at the rub; narrow it, or file more than one record" ) lines = path.read_text().splitlines() if end > len(lines): raise ValueError(f"{path} has {len(lines)} lines, so {start}-{end} does not exist") width = len(str(end)) body = "\n".join(f"{n:>{width}} {lines[n - 1]}" for n in range(start, end + 1)) where = path.resolve().relative_to(Path(root).resolve()) if root else path return f"{where}:{start}-{end}", body PK\442?2?plumb/launch.py"""The launcher: a project's declaration in, a manifest out, whichever language the artifact is. ``plumb`` is not a Python tool that grew a Java mode. It is a **launcher over a language-neutral core** (``ch0-4``): it reads the one config, runs the adapter that config names, and hands the resulting manifest to the gate. A third language is a new branch here and nothing else — the gate, the board and the manifest do not change, and none of them learns a language. Everything an adapter needs arrives **resolved**. The Python adapter is called in-process with a :class:`ProductionSurface`; the Java adapter is handed a run file of already-decided facts and parses no configuration format at all. That is what keeps one config file honest across two languages: there is one reader, and it is here. This module knows adapter names, which is exactly why it is not in :mod:`plumb.core`. """ import os import shutil import subprocess import sys import time from pathlib import Path from typing import Callable from plumb import runfile from plumb.config import JAVA, PYTHON, Config, InvalidConfig from plumb.core.manifest import Manifest #: The resolved facts handed to a non-Python adapter. One `kindvalue…` per line, because an #: adapter reading its own configuration is an adapter that has to agree with every other one #: about a format — and the Java side hand-writes its JSON precisely to avoid taking on parsers. RUN_FILE = "plumb-run.tsv" #: How long to wait for a debuggee to announce its port before giving up on grounding. LISTEN_TIMEOUT = 30.0 class LaunchFailed(RuntimeError): """The adapter could not be run. Distinct from a manifest saying a story failed: this is the tool being unable to ask, and it must never read as an answer.""" def produce(config: Config, *, pytest_args: list[str] | None = None, mutation: bool = False, report: Callable[[str], None] | None = None) -> Manifest: """Run the adapter the project declared, and return the manifest it produced.""" if config.adapter == JAVA: return _java(config, mutation=mutation or config.mutation, report=report) if config.adapter == PYTHON: return _python(config, pytest_args or [], mutation=mutation or config.mutation, report=report) raise LaunchFailed(f"no adapter named {config.adapter!r}") def _python(config: Config, pytest_args: list[str], *, mutation: bool, report: Callable[[str], None] | None) -> Manifest: """In this interpreter when it is the project's, and in the project's when it is not. The adapter imports the project's code and its test framework, so it has to run where those live. While Plumb is a checkout on the path that is the same interpreter and nothing else happens; installed as its own artifact it is emphatically not, and running the adapter in the tool's interpreter would fail to import the project — or worse, import a *different* copy of it that happens to be on the path. Staying in-process where possible is not only speed: the tracer sees frames in the process it is installed in, so a caller driving the library directly keeps working exactly as before. """ interpreter = _interpreter(config) if interpreter == Path(sys.executable): from plumb.adapters.python.adapter import collect from plumb.adapters.python.mutation import OptIn from plumb.adapters.python.surface import ProductionSurface return collect(pytest_args, surface=ProductionSurface.from_config(config), mutation=OptIn(enabled=mutation), report=report, marker=config.marker) say = report or (lambda line: None) say = report or (lambda line: None) work = (config.root / "build" / "plumb").resolve() manifest_path = work / "manifest.json" run_file = runfile.write(config, work / RUN_FILE, pytest_args=tuple(pytest_args), mutation=mutation) say(f"python: {interpreter}") completed = subprocess.run( [str(interpreter), "-m", "plumb.adapters.python", str(run_file), str(manifest_path)], cwd=config.root, env={**os.environ, "PYTHONPATH": _own_path()}, text=True, # The child's chatter goes to stderr, explicitly. `contextlib.redirect_stdout` rebinds # only Python's `sys.stdout`; a subprocess inherits the OS-level descriptor and writes # straight onto the channel `--json` is emitted on. That corrupts the neutral format # (`ch5-2`) for every project with its own interpreter — the normal case, and exactly the # path in-process running never exercises. stdout=sys.stderr) if completed.returncode != 0 or not manifest_path.is_file(): raise LaunchFailed( f"the Python adapter failed under {interpreter} (exit {completed.returncode})") return Manifest.read(manifest_path) def _interpreter(config: Config) -> Path: """The interpreter the *project* runs under, which is not necessarily the tool's. Declared beats discovered beats current. The discovery is deliberately dumb — a venv in the usual place, or one already active — because guessing harder is how a tool picks the wrong Python and reports a project's own code missing. """ declared = config.options.get("interpreter") if config.adapter == PYTHON else None if declared: return _absolute(config.root / declared) active = os.environ.get("VIRTUAL_ENV") candidates = [Path(active) / "bin" / "python"] if active else [] candidates += [config.root / ".venv" / "bin" / "python", config.root / "venv" / "bin" / "python"] for candidate in candidates: if candidate.is_file(): return _absolute(candidate) return Path(sys.executable) def _absolute(path: Path) -> Path: """Absolute, and deliberately **not** resolved. A virtualenv's ``bin/python`` is a symlink to the base interpreter, and Python works out ``sys.prefix`` from the path it was *invoked* by. Following the link therefore runs the base interpreter with none of the venv's packages — the project's own test framework included. This passed a first test anyway, because the machine happened to have pytest installed globally too: the venv was destroyed and the run succeeded on the wrong interpreter's packages. A resolved path is the kind of tidiness that silently answers a different question. """ return Path(os.path.abspath(path)) def _own_path() -> str: """Where this code lives, so a subprocess under another interpreter can import it. A zipapp is a zip, and Python imports out of a zip on ``sys.path`` — so the artifact adds itself to ``PYTHONPATH`` and needs nothing installed into the project's environment. That only works because the core has no dependencies: there is nothing else to bring along. """ here = Path(__file__).resolve() for parent in here.parents: # Inside a zipapp the package sits under the archive itself, which is a file, not a dir. if parent.is_file(): return str(parent) return str(here.parent.parent) # ---- Java --------------------------------------------------------------------------------- def _java(config: Config, *, mutation: bool, report: Callable[[str], None] | None) -> Manifest: """Three processes, orchestrated here rather than in a shell script. The script this replaces had to be run by hand, in the right order, with its own copy of the classpath — which is the *"run two shell scripts, then point plumb at a file"* that made Java adoption cost a reading of this repository. Grounding needs the suite under JDWP with the observer attached, and the observer needs the port the suite chose. That handshake is the reason this cannot simply be one command. """ say = report or (lambda line: None) options = config.options work = (config.root / options.get("work", "build/plumb")).resolve() work.mkdir(parents=True, exist_ok=True) manifest_path = work / "manifest.json" java = shutil.which("java") or "java" classpath = _classpath(config) runfile.write(config, work / RUN_FILE, mutation=mutation) tests = options.get("tests") if not tests: raise InvalidConfig("plumb.toml: [java] needs `tests` — the classpath root holding the " "compiled test classes, which is what the runner discovers from") tests_root = (config.root / tests).resolve() observed = config.declares_anything port = _free_port() if observed else None command = [java, "-ea"] if observed: command.append(f"-agentlib:jdwp=transport=dt_socket,server=y,suspend=y," f"address=127.0.0.1:{port}") command.append("-Dplumb.observed=true") command += ["-cp", classpath, "plumb.PlumbRunner", str(tests_root), str(manifest_path), str(work / RUN_FILE)] out = (work / "run.out").open("w") suite = subprocess.Popen(command, cwd=config.root, stdout=out, stderr=subprocess.STDOUT) try: if observed: _await_listening(work / "run.out", suite) say(f"java: observing on port {port}") _run(java, ["--add-modules", "jdk.jdi", "-cp", classpath, "plumb.Observer", str(port), str(work / RUN_FILE), str(manifest_path), *( [str(work / "wired.txt")] if mutation else [] )], cwd=config.root, say=say, what="observer") else: suite.wait(timeout=600) finally: out.close() # The suite JVM outlives its own run whenever the project leaves a non-daemon thread # alive, which real suites do constantly. Waiting on it would hang the command. suite.terminate() if mutation: _run(java, ["-cp", classpath, "plumb.Mutate", str((config.root / (config.sources[0] if config.sources else ".")).resolve()), str(tests_root), classpath, str(work)], cwd=config.root, say=say, what="mutation", stream=True) if not manifest_path.is_file(): log = (work / "run.out").read_text() if (work / "run.out").is_file() else "" if "org/junit/platform/launcher" in log: # Surefire supplies the launcher internally, so an ordinary Maven project never # declares it and `dependency:build-classpath` — the command this tool tells people # to run — leaves it out. The failure surfaces as a class the adopter never heard of. raise LaunchFailed( "the Java adapter needs junit-platform-launcher on the classpath, and your build " "does not put it there. Maven's Surefire supplies it internally, so it is absent " "from `dependency:build-classpath`. Add it as a test dependency and rebuild the " "classpath file:\n\n" " \n" " org.junit.platform\n" " junit-platform-launcher\n" " test\n" " ") raise LaunchFailed( f"the Java adapter produced no manifest at {manifest_path} — see {work / 'run.out'}") return Manifest.read(manifest_path) def _classpath(config: Config) -> str: """The project's own test classpath, plus wherever Plumb's Java classes live. The adapter's own jar is appended by the tool, never configured: it ships *inside* the artifact, so where it lives is Plumb's business and not a setting anybody has to keep correct. ``plumb_classpath`` remains only for a checkout driving compiled classes directly. """ from_file = "" if config.options.get("classpath_file"): # What a build tool can actually produce — `mvn dependency:build-classpath` writes one, # and a classpath pasted into config goes stale the first time a dependency moves. path = (config.root / config.options["classpath_file"]).resolve() if not path.is_file(): raise InvalidConfig(f"plumb.toml: [java] classpath_file {path} does not exist — " "the project has not been built yet, or the path is wrong") from_file = path.read_text().strip() parts = [config.options.get("classpath", ""), from_file, os.environ.get("PLUMB_JAVA_CLASSPATH", ""), config.options.get("plumb_classpath", ""), _bundled_jar()] joined = ":".join(p for p in parts if p) if not joined: raise InvalidConfig("plumb.toml: [java] needs `classpath` — the project's compiled " "classes and its test framework, as `java -cp` would take them") return joined def _bundled_jar() -> str: """The Java adapter that shipped with this tool, made reachable to a `java -cp`. Inside the artifact the jar is a member of a zip, and a JVM cannot read it there — so it is extracted once, to a cache keyed by version. Keyed by version because the alternative is an upgrade that silently keeps running the previous adapter, which is a wrong answer rather than a stale one. Never configured. The jar ships *inside* the artifact, so where it lives is Plumb's business and not a setting an adopter has to keep correct. """ from plumb import __version__ here = Path(__file__).resolve() archive = next((parent for parent in here.parents if parent.is_file()), None) if archive is None: # A checkout keeps the jar where the build script leaves it, which is NOT where the # artifact carries it. Looking only at the packaged location made this return nothing # from a source tree — and the caller then did nothing, quietly. for candidate in (here.parent / "adapters" / "java" / "plumb-java.jar", here.parents[2] / "adapters" / "java" / "build" / "plumb-java.jar"): if candidate.is_file(): return str(candidate) return "" import zipfile cached = Path(os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache")) / "plumb" / __version__ jar = cached / "plumb-java.jar" if not jar.is_file(): member = "plumb/adapters/java/plumb-java.jar" with zipfile.ZipFile(archive) as bundle: if member not in bundle.namelist(): return "" cached.mkdir(parents=True, exist_ok=True) jar.write_bytes(bundle.read(member)) return str(jar) def _await_listening(log: Path, suite: subprocess.Popen) -> None: deadline = time.monotonic() + LISTEN_TIMEOUT while time.monotonic() < deadline: if suite.poll() is not None: raise LaunchFailed(f"the Java suite exited before it began listening — see {log}") if log.is_file() and "Listening" in log.read_text(): return time.sleep(0.05) raise LaunchFailed(f"the Java suite never announced its debug port — see {log}") def _run(java: str, args: list[str], *, cwd: Path, say, what: str, stream: bool = False) -> None: # `stream` still means "let it through as it happens", but onto stderr — never stdout, which # is the channel the board is emitted on. completed = subprocess.run([java, *args], cwd=cwd, text=True, capture_output=not stream, **({} if not stream else {"stdout": sys.stderr})) if not stream and completed.stdout: for line in completed.stdout.splitlines(): say(f"java: {line}") if completed.returncode != 0: # Never silent. A failed observer means grounding is `not-checked` rather than wrong, and # a reader has to be told which of those they are looking at (`ch3-7`). say(f"java: {what} failed ({completed.returncode}) — the board reports what it could get") def _free_port() -> int: import socket with socket.socket() as probe: probe.bind(("127.0.0.1", 0)) return probe.getsockname()[1] PK \tzY#Y#plumb/receiver.py"""The receiving end of the support API — where an enrolled contributor's records land. Deliberately stdlib-only. It runs on a box already serving production sites, so the smallest possible thing that can be correct is the right thing: no framework, no dependency tree to keep patched, no build step. It binds to localhost and expects a reverse proxy in front for TLS. It is the **second** API (``ch6-1``): it touches no manifest, fills no slot, and moves no story's status. Nothing it accepts can become evidence. ``ch6-5`` lives here, and only here: the client states a relationship, and this overwrites it from the credential it actually authenticated. Identity the filer types cannot be trusted, so the one place it can be established is the boundary that checked a secret. """ import hmac import json import re from dataclasses import dataclass from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from plumb.friction import CONDITIONAL, REQUIRED, FrictionRecord, InvalidRecord from plumb.transmit import VERSION_HEADER MAX_BODY_BYTES = 256 * 1024 SAFE_ID = re.compile(r"\A[0-9a-z][0-9a-z-]{0,120}\Z") from plumb.registry_ops import fingerprint @dataclass(frozen=True) class Contributor: label: str relationship: str class Registry: """Credential → contributor. A flat file, because the whole population is people you enrolled by hand and it is never going to be large enough to need anything else.""" def __init__(self, path: Path | str): self.path = Path(path) self._by_credential = self._load() def _load(self) -> dict[str, Contributor]: if not self.path.is_file(): return {} raw = json.loads(self.path.read_text()) # Keyed by FINGERPRINT, not by the credential. A registry holding credentials in the # clear is one that can redisplay them, and `ch9-9` is explicit that a store which can do # that is a store worth stealing. A pre-hash entry is still honoured — hashed on read — # so an existing deployment keeps authenticating while it is migrated. return { entry.get("credential_sha256") or fingerprint(entry["credential"]): Contributor(entry.get("label", "unnamed"), entry["relationship"]) for entry in raw.get("contributors", []) } def resolve(self, presented: str | None) -> Contributor | None: """Constant-time across every known credential, so timing cannot enumerate them.""" if not presented: return None offered = fingerprint(presented) found = None for known, contributor in self._by_credential.items(): if hmac.compare_digest(known, offered): found = contributor return found #: The record shapes this receiver understands (``ch6-13``). A set, for the same reason the #: manifest's is: older is readable, newer cannot be guessed at. ACCEPTED_VERSIONS = frozenset({"1"}) def specification() -> dict: """What this receiver accepts, served from the receiver itself (``ch6-14``). Derived from the record definition rather than written beside it, so it describes the receiver that is *running* rather than the one somebody documented — and when those two disagree, the running one is the one refusing records. """ return { "service": "plumb-friction", "record_versions": sorted(int(v) for v in ACCEPTED_VERSIONS), "version_header": VERSION_HEADER, "endpoints": { "POST /friction": "submit one record; requires a bearer credential", "GET /spec": "this document", "GET /health": "liveness", }, "record": {"required": list(REQUIRED), "conditional": list(CONDITIONAL)}, "refusals": { "401": "credential not recognised — retryable once re-enrolled", "413": "body too large — terminal", "422": "content refused; see `error` — terminal", "426": "record version not accepted; see `accepts` — retryable after upgrade", }, } class FrictionHandler(BaseHTTPRequestHandler): server_version = "plumb-friction" sys_version = "" # do not advertise the Python version to the internet registry: Registry store_root: Path def do_GET(self): if self.path == "/health": self._respond(200, {"status": "ok"}) elif self.path.rstrip("/") == "/spec": self._respond(200, specification()) else: self._respond(404, {"error": "no such endpoint"}) def do_POST(self): if self.path.rstrip("/") != "/friction": return self._respond(404, {"error": "no such endpoint"}) contributor = self.registry.resolve(self._presented_credential()) if contributor is None: return self._respond(401, {"error": "unrecognised credential"}) declared = self.headers.get(VERSION_HEADER) if declared is not None and declared.strip() not in ACCEPTED_VERSIONS: # ch6-13: name what is spoken here. A refusal that says only "not that" leaves the # producer guessing, and the client retries this rather than discarding the record. return self._respond(426, { "error": f"record version {declared!r} is not accepted here", "accepts": sorted(int(v) for v in ACCEPTED_VERSIONS), }) body = self._read_body() if body is None: return self._respond(413, {"error": f"body exceeds {MAX_BODY_BYTES} bytes"}) try: record = FrictionRecord.from_json(body.decode("utf-8")) except (InvalidRecord, ValueError, UnicodeDecodeError) as rejected: # ch6-4: name what is wrong, so a writer still holding the context can fix it. # 422 rather than 400: this is the content, positively identified as unusable, and # the client treats it as terminal (`ch6-13`). A bare 400 cannot be told apart from # a version the receiver has not learned yet, so it must never mean this. return self._respond(422, {"error": str(rejected), "refused": "content"}) if not SAFE_ID.match(record.id): return self._respond(422, {"error": "id must be lowercase alphanumeric and hyphens", "refused": "content"}) # ch6-5: the credential decides who this is, never the payload. stamped = FrictionRecord(**{**record.__dict__, "source": contributor.relationship}) destination = self.store_root / f"{stamped.id}.json" if destination.exists(): return self._respond(409, {"status": "already held", "id": stamped.id}) self.store_root.mkdir(parents=True, exist_ok=True) destination.write_text(stamped.to_json()) self._respond(201, {"status": "stored", "id": stamped.id, "source": stamped.source}) def _presented_credential(self) -> str | None: header = self.headers.get("Authorization", "") return header[7:].strip() if header.lower().startswith("bearer ") else None def _read_body(self) -> bytes | None: length = int(self.headers.get("Content-Length") or 0) return None if length > MAX_BODY_BYTES else self.rfile.read(length) def _respond(self, status: int, payload: dict) -> None: body = (json.dumps(payload) + "\n").encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, fmt, *args): """One line per request, to stderr, for journald. Never the body: a record is a contributor's words and its anchors, and a log is not a store. Written directly rather than via log_error, which routes back here.""" import sys as _sys print(f"{self.address_string()} {fmt % args}", file=_sys.stderr, flush=True) def build(store_root: Path | str, registry_path: Path | str, host="127.0.0.1", port=8081): handler = type("BoundFrictionHandler", (FrictionHandler,), { "registry": Registry(registry_path), "store_root": Path(store_root), }) return ThreadingHTTPServer((host, port), handler) def main(argv=None) -> int: import argparse parser = argparse.ArgumentParser(prog="plumb-friction-receiver", description=__doc__) parser.add_argument("--store", required=True, help="directory records are written to") parser.add_argument("--registry", required=True, help="JSON file of enrolled contributors") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8081) args = parser.parse_args(argv) server = build(args.store, args.registry, args.host, args.port) print(f"listening on {args.host}:{args.port}, storing to {args.store}", flush=True) server.serve_forever() return 0 PK \o0plumb/registry_ops.py"""``ch9-9`` — a credential is issued, listed and revoked by an **operation**, never a hand-edit. Admitting a contributor is what ``ch6-5`` makes load-bearing: the credential is the only thing that establishes who a record came from, because the filer's own claim about their identity is exactly what cannot be trusted. So minting one is a command against the registry, not somebody editing JSON on a live host at two in the morning. Three properties, and each was absent before: * **Shown exactly once.** A store that can redisplay a credential is a store worth stealing, so what is kept is a SHA-256 of it and never the credential itself. Losing one means minting another, which is the correct cost. * **The listing names who, never what.** "Who has access" is a reasonable question; it is not a request to be handed every secret in the file. * **Revocation exists at all.** That is what makes ``ch6-6``'s consent revocable in the direction the *receiver* controls, as ``ch6-9`` makes it revocable in the direction the contributor does. """ import hashlib import json import secrets from dataclasses import dataclass from pathlib import Path #: Long enough that guessing is not a strategy, and URL-safe so it survives being pasted into a #: config file, a header and a terminal without anything helpfully re-encoding it. CREDENTIAL_BYTES = 32 class RegistryError(ValueError): """Refused, naming what is already true. Every one of these is read by an operator mid-task.""" @dataclass(frozen=True) class Entry: label: str relationship: str credential_sha256: str def fingerprint(credential: str) -> str: return hashlib.sha256(credential.encode()).hexdigest() def _read(path: Path) -> list[dict]: if not path.is_file(): return [] return json.loads(path.read_text()).get("contributors", []) def _write(path: Path, contributors: list[dict]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps({"contributors": contributors}, indent=2, sort_keys=True) + "\n") def mint(path: Path | str, label: str, relationship: str = "trusted-tester") -> str: """Issue a credential and return it **once**. The registry keeps only its fingerprint. The label must be unique: two contributors sharing one is a registry that cannot answer the question revocation asks — *which of them am I removing?* """ path = Path(path) if not label.strip(): raise RegistryError("a contributor needs a label — it is what revocation names later") contributors = _read(path) if any(entry.get("label") == label for entry in contributors): raise RegistryError( f"{label!r} is already enrolled. Revoke that one first if you are replacing it — " "two entries with one label cannot be told apart when you come to remove one") credential = secrets.token_urlsafe(CREDENTIAL_BYTES) contributors.append({"label": label, "relationship": relationship, "credential_sha256": fingerprint(credential)}) _write(path, contributors) return credential def entries(path: Path | str) -> list[Entry]: """Who is enrolled. Deliberately returns no secret, so no caller can leak one by accident.""" return [ Entry(label=raw.get("label", "unnamed"), relationship=raw.get("relationship", "unknown"), # A pre-hash entry keeps working; what it cannot do is be displayed, which is the # property that matters. Migration is: revoke it and mint another. credential_sha256=raw.get("credential_sha256") or (fingerprint(raw["credential"]) if raw.get("credential") else "unknown")) for raw in _read(Path(path)) ] def revoke(path: Path | str, label: str) -> bool: """Remove a contributor. Returns whether there was one to remove. Idempotent on purpose: an operator revoking access should never be left wondering whether it worked because the second attempt errored. """ path = Path(path) contributors = _read(path) remaining = [entry for entry in contributors if entry.get("label") != label] if len(remaining) == len(contributors): return False _write(path, remaining) return True PK \3  plumb/runfile.py"""The launcher's resolved facts: written here, read by whichever adapter was chosen. ``kindvalue…`` per line, and deliberately not a config format. The launcher has already decided everything — which entry points, which production code, which stories are structural — so what crosses this boundary is answers, not questions. An adapter that needs a parser has been given the wrong job, which is why the Java side's reader is one ``split("\\t")``. It exists because **an adapter may not run in this process**. Java never does; Python does not either once the tool is installed as an artifact and the project has its own interpreter. A file is what survives that boundary, and using the same one for both keeps the two adapters honest about being the same shape. """ from pathlib import Path from plumb import __version__ ENTRY_POINT, PRODUCTION, ARCHETYPE, PYTEST_ARG, MUTATION, VERSION = ( "entrypoint", "production", "archetype", "pytest", "mutation", "version") MARKER = "marker" def write(config, path: Path, *, pytest_args: tuple[str, ...] = (), mutation: bool = False) -> Path: """Resolve everything an adapter could need and write it down. Archetypes are included because resolving them is the **core's** job (``ch2-7``): they are declared on the story in the spec, and an adapter keeping its own copy is a second home for something one home owns. A run that executed the spec has been told them directly; anything else takes them from config, which is the interim the archetype question has not settled. """ from plumb.adapters.python import archetypes resolved = {**config.archetypes, **archetypes.declared()} lines = [f"{ENTRY_POINT}\t{e}" for e in config.entry_points] lines += [f"{PRODUCTION}\t{p}" for p in config.production] lines += [f"{ARCHETYPE}\t{story}\t{kind}" for story, kind in sorted(resolved.items())] lines += [f"{PYTEST_ARG}\t{a}" for a in pytest_args] lines.append(f"{MUTATION}\t{'true' if mutation else 'false'}") # The tool's own version, so an adapter never carries a second copy of it. The jar ships # inside the artifact, so a hardcoded version there is a claim about *this* tool made by a # file that cannot see it — and it goes stale on the first release that forgets to edit it. lines.append(f"{VERSION}\t{__version__}") lines.append(f"{MARKER}\t{config.marker}") path.parent.mkdir(parents=True, exist_ok=True) path.write_text("\n".join(lines) + "\n") return path def read(path: Path) -> dict[str, list[list[str]]]: """Facts by kind. A missing file reads as nothing declared, never as an error (``ch0-7``).""" found: dict[str, list[list[str]]] = {} path = Path(path) if not path.is_file(): return found for line in path.read_text().splitlines(): fields = line.split("\t") if len(fields) > 1: found.setdefault(fields[0], []).append(fields[1:]) return found def values(facts: dict[str, list[list[str]]], kind: str) -> tuple[str, ...]: return tuple(fact[0] for fact in facts.get(kind, ())) PK\QN+p.p.plumb/scaffold.py"""``plumb init`` — get a project from nothing to a board it can run (``ch9-5``, ``ch9-7``). Adoption was measured and found to cost a reading of this repository. The expensive part was never the concepts; it was working out what to *write down* — which classes count as production, where the compiled tests live, what a classpath file is for. This detects what it can, writes the one config file, and says what to do next in commands rather than prose. **It completes, and never overwrites** (``ch9-5``). What is missing is written, what is already there is left exactly as it is, and both are reported. Adopting in pieces is the observed failure mode, so finishing a partial adoption is the common path — and a tool that destroys work when run twice is a tool nobody dares run once. What this does **not** yet do is emit a spec: the claim vocabulary, a first chapter paired with its flow, and the integrity gates (``ch9-1``, ``ch9-2``, ``ch9-3``). A project does not need one to get a board — story ids are opaque and Plumb keeps no catalog (``ch2-2``) — so this is the smaller, honest half, and those stories stay unproven rather than being claimed by a citation that would not support them. """ from dataclasses import dataclass, replace from pathlib import Path from plumb.config import CONFIG_NAME, JAVA, PYTHON @dataclass(frozen=True) class Detected: """What the project looks like, and what it would therefore be told to declare.""" adapter: str production: tuple[str, ...] = () sources: tuple[str, ...] = () entry_points: tuple[str, ...] = () options: dict = None notes: tuple[str, ...] = () def detect(root: Path) -> Detected: """Guess from the layout, and be obvious about what is a guess. Never silent: every value this cannot work out is emitted as a commented placeholder with the question spelled out, because a config file that looks complete and is wrong costs more than one that is visibly unfinished. """ if (root / "pom.xml").is_file() or (root / "build.gradle").is_file() \ or (root / "build.gradle.kts").is_file(): return _java(root) return _python(root) def _java(root: Path) -> Detected: maven = (root / "pom.xml").is_file() tests = "target/test-classes" if maven else "build/classes/java/test" production = tuple(f"{p}.*" for p in _java_packages(root / "src" / "main" / "java")) \ or ("com.example.*",) packages = [p for p in production if p != "com.example.*"] notes = () if packages else ( "no src/main/java found — `production` is a placeholder, set it to your own packages",) return Detected( adapter=JAVA, production=production, sources=("src/main/java",), entry_points=(), options={"tests": tests, "classpath_file": "target/plumb-classpath.txt" if maven else "build/plumb-classpath.txt"}, notes=notes + ( "entry_points is empty: nothing can be checked for reaching the wired system until " "you name at least one", "classpath_file must exist before `plumb board` runs — with Maven: " "mvn dependency:build-classpath -Dmdep.outputFile=target/plumb-classpath.txt", "add junit-platform-launcher as a test dependency — Surefire supplies it internally, " "so it is missing from the dependency classpath and the adapter cannot start without it", ), ) def _java_packages(source_dir: Path) -> list[str]: """The deepest package prefix that still covers everything. Stopping at the first directory would emit `com.*`, which matches every library in the classpath as well as the project — and the observer filters production classes by that pattern, so an over-broad one turns a narrow probe into a whole-program trace. Descend while there is exactly one way down and no source yet. """ if not source_dir.is_dir(): return [] found = [] for top in sorted(p for p in source_dir.iterdir() if p.is_dir()): node, parts = top, [top.name] while True: children = [c for c in node.iterdir() if c.is_dir()] if list(node.glob("*.java")) or len(children) != 1: break node = children[0] parts.append(node.name) found.append(".".join(parts)) return found def _python(root: Path) -> Detected: src = root / "src" candidates = [p for p in (src.iterdir() if src.is_dir() else []) if (p / "__init__.py").is_file()] if not candidates: candidates = [p for p in root.iterdir() if p.is_dir() and (p / "__init__.py").is_file() and not p.name.startswith(".")] production = tuple(str(p.relative_to(root)) for p in sorted(candidates)) or ("src/myapp",) return Detected( adapter=PYTHON, production=production, sources=production, entry_points=(), options={}, notes=(() if candidates else ("no package found — `production` is a placeholder",)) + ( "entry_points is empty: nothing can be checked for reaching the wired system until " "you name at least one", ), ) def render(found: Detected) -> str: """The config file, with the reasoning inline. Comments are the cheapest documentation there is: they arrive with the thing they explain and cannot be looked for and not found.""" lines = [ "# What this project declares about itself. One file, read by Plumb for every language.", f'adapter = "{found.adapter}"', "", "# What counts as production code, in your language's own spelling.", f"production = {_toml_list(found.production)}", "# Where that code's source lives — mutation needs the files, not the compiled form.", f"sources = {_toml_list(found.sources)}", "", "# Where the wired system begins: the handful of places a real request enters.", "# A story is only checked for reaching the wired system if at least one is named.", f"entry_points = {_toml_list(found.entry_points)}", "", "# Off because it is slow, not because it is optional. `plumb board --mutation` for a run.", "mutation = false", ] if found.options: lines += ["", f"[{found.adapter}]"] lines += [f'{key:<14} = "{value}"' for key, value in sorted(found.options.items())] if found.notes: lines += ["", "# Still to do:"] + [f"# - {note}" for note in found.notes] return "\n".join(lines) + "\n" def _toml_list(values) -> str: return "[" + ", ".join(f'"{v}"' for v in values) + "]" @dataclass(frozen=True) class Outcome: written: tuple[str, ...] kept: tuple[str, ...] detected: Detected @property def anything_written(self) -> bool: return bool(self.written) #: The spec machinery, identical in every project that uses this mechanism (`ch9-2`). SPEC_FILES = ("status.py", "test_spec_backlog.py", "test_spec_status_guardrail.py") def _machinery(name: str) -> str | None: """The one text, from wherever this copy of Plumb is running. Packaged inside the artifact, or in `spec/` in a checkout — and the SAME bytes either way, which is the whole of `ch9-2`: what this repository runs on itself is a projection of the template, not a sibling of it. """ here = Path(__file__).resolve() for candidate in (here.parent / "templates" / name, here.parents[2] / "spec" / name): if candidate.is_file(): return candidate.read_text() return None #: Where a Java project keeps the annotation jar. Inside the project, because a compile classpath #: is a project-relative thing and a path into somebody's home directory does not survive being #: committed, shared, or run in CI. JAR_IN_PROJECT = Path(".plumb") / "plumb-java.jar" def init(root: Path | str = ".", spec: bool = False) -> Outcome: """Write what is missing; keep what is there (``ch9-5``).""" root = Path(root) found = detect(root) written, kept = [], [] config = root / CONFIG_NAME if config.exists(): kept.append(CONFIG_NAME) else: config.write_text(render(found)) written.append(CONFIG_NAME) if found.adapter == JAVA: # Without this the adopter cannot compile a citation at all: `@Proves` needs the jar on # the COMPILE classpath, and it otherwise appears only as a side effect of a successful # run — which cannot happen until something is cited. Handing it over here is what breaks # that circle (`ch9-8`). jar = root / JAR_IN_PROJECT if jar.exists(): kept.append(str(JAR_IN_PROJECT)) elif _place_jar(jar): written.append(str(JAR_IN_PROJECT)) else: # Never silent. Without the jar the adopter cannot compile a citation at all, so a # quiet failure here leaves them stuck with nothing to look at. found = replace(found, notes=found.notes + ( "could not find the Java adapter jar to copy into .plumb/ — citations will not " "compile until it is there; reinstall the tool, or copy it from a checkout's " "adapters/java/build/plumb-java.jar",)) if spec: from plumb.spec_scaffold import emit made, left = emit(root) written += made kept += left return Outcome(written=tuple(written), kept=tuple(kept), detected=found) def _place_jar(target: Path) -> bool: """Copy the adapter jar out of whatever Plumb is running from.""" from plumb.launch import _bundled_jar source = _bundled_jar() if not source: return False target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(Path(source).read_bytes()) return True #: The next action, in commands (``ch9-7``). A capability nobody is told to reach for is a #: capability nobody uses, and "handed a directory" is not the same as "handed a method". _CITE = { JAVA: (' import plumb.Proves;\n\n' ' @Test\n' ' @Proves(value = "APP-1", depth = "wiring")\n' ' void ordersAreRouted() { ... }'), PYTHON: (' import pytest\n\n' ' @pytest.mark.proves("APP-1", depth="wiring")\n' ' def test_orders_are_routed(): ...'), } def next_steps(found: Detected) -> str: lines = ["", "Next:", "", " 1. Name where your wired system begins, in plumb.toml:", "", ' entry_points = ["com.myapp.Api#handleRequest"]' if found.adapter == JAVA else ' entry_points = ["myapp.cli:main"]', "", " 2. Cite a story from a test. The id is yours — Plumb keeps no catalog of them:", "", _CITE[found.adapter], "", " 3. Run it:", "", " plumb board", ""] if found.adapter == JAVA: lines += ["`@Proves` lives in the jar written to .plumb/plumb-java.jar. Depend on it to", "compile your citations — Maven:", "", " ", " plumb", " plumb-java", " 0", " system", " ${project.basedir}/.plumb/plumb-java.jar", " ", "", "…or Gradle:", "", " testImplementation files('.plumb/plumb-java.jar')", "", "Then write your test classpath to a file, which your build tool already does:", "", " mvn dependency:build-classpath -Dmdep.outputFile=target/plumb-classpath.txt", ""] return "\n".join(lines) PKG\bplumb/spec_scaffold.py"""Emit a runnable spec from the same text Plumb runs on itself (``ch9-1``, ``ch9-2``, ``ch9-3``). The machinery — the claim vocabulary, the backlog, the integrity gates — is copied **verbatim**. That is the whole of ``ch9-2``: what a new project receives and what this repository's own ``spec/`` contains are one artifact, held equal by a check, so a divergence is a red build rather than a documentation chore. The alternative is the state chapter 9 was written from, a reference implementation *acting* as a template, which drifts by default because nothing fails when it does. Only ``conventions.py`` is rendered, because the id namespace and the file naming are the whole of what differs between two specs built this way (``ch9-4``). The gates go with it (``ch9-3``). The evidence for that is a real adopter who copied the backlog printer and not the gates, and so built a spec that could rot without anything going red — a gate that ships only to the tool's author protects only the tool's author. """ from pathlib import Path #: Carried verbatim. Adding one here is the only thing needed to ship it to every adopter. MACHINERY = ("status.py", "test_spec_backlog.py", "test_spec_status_guardrail.py") class ScaffoldUnavailable(RuntimeError): """This build cannot emit a spec that would run, so it emits none.""" FLOW_DIR = "diagrams" FIRST_MODULE = "chapter_01_first.py" FIRST_FLOW = "chapter-01-first.md" CONVENTIONS = '''"""The only thing that differs between two specs built by this mechanism (ch9-4). Everything beside this file — the claim vocabulary, the backlog printer, the integrity gates — is identical in every project that uses it, which is what lets them be one artifact rather than a template and a copy that drift apart. """ #: The prefix every story id carries. NAMESPACE = "{ns}" #: How a chapter module is named. Its stem pairs with a flow of the same number. MODULE_GLOB = "chapter_*.py" #: Where the process flows live, relative to this directory. FLOW_DIR = "diagrams" #: How a flow file is named. FLOW_GLOB = "chapter-*.md" #: What a citing test looks like. CITATION_MARKER = "proves" ''' CHAPTER = '''"""Chapter 1 — your first chapter. STATUS IS DERIVED — the ideal and its process flow live in ``{flow_dir}/{flow}``. Replace this with what your project actually claims. A story declares only that it exists; whether it is proven is computed from a test that cites it, ran, and passed. """ from status import story def {ns}1_1(): story("{ns}1-1") ''' FLOW = '''# Chapter 1 — your first chapter What this chapter claims, and how work flows through it. Spec source: `../{module}`. ```mermaid flowchart TD START([START the input this chapter begins with]):::terminal P1[{ns}1-1 the first thing that happens]:::process DONE([DONE what this chapter produces]):::terminal START --> P1 --> DONE classDef terminal fill:#eef0ee,stroke:#52514e,color:#0b0b0b; classDef process fill:#e9eef5,stroke:#2a78d6,color:#0b0b0b; ``` ## Story → test trace | Story | What it proves | |---|---| | `{ns}1-1` | **Replace this with a real claim** - one sentence saying what must be true, and why it matters. A story nobody can state plainly is a story nobody can prove | ## Input → process → output **Input** - what this chapter is handed **Process** - **[P1]** the first thing that happens (`{ns}1-1`) **Output** - what it produces ## Open unknowns - none yet. An open question belongs here, anchored to the story it affects. ''' #: Added to the project's pytest configuration, because a spec that the runner does not discover #: is a directory of files rather than something that runs (``ch9-1``). PYTEST_HINT = '''[tool.pytest.ini_options] testpaths = ["spec", "tests"] pythonpath = ["spec"] python_files = ["chapter_*.py", "test_*.py"] python_functions = ["test_*", "{ns}[0-9]*_[0-9]*"] markers = ["proves(id, depth, ref): cite the story this test proves"] ''' def machinery(name: str) -> str | None: """The one text, from wherever this copy of Plumb is running — packaged inside the artifact, or ``spec/`` in a checkout, and the same bytes either way. A path *inside* a zipapp is not a file the OS can see, so `is_file()` is False for every one of them and a plain filesystem read finds nothing. That is why the archive is opened explicitly, exactly as the Java jar is. """ here = Path(__file__).resolve() archive = next((parent for parent in here.parents if parent.is_file()), None) if archive is not None: import zipfile with zipfile.ZipFile(archive) as bundle: member = f"plumb/templates/{name}" if member in bundle.namelist(): return bundle.read(member).decode() return None for candidate in (here.parent / "templates" / name, here.parents[2] / "spec" / name): if candidate.is_file(): return candidate.read_text() return None def emit(root: Path, namespace: str = "ch") -> tuple[list[str], list[str]]: """Write a spec that runs. Never overwrites; reports written and kept (``ch9-5``).""" spec = Path(root) / "spec" files = {name: machinery(name) for name in MACHINERY} files["conventions.py"] = CONVENTIONS.format(ns=namespace) files[FIRST_MODULE] = CHAPTER.format(flow_dir=FLOW_DIR, flow=FIRST_FLOW, ns=namespace) missing = [name for name, text in files.items() if text is None] if missing: # Never a partial spec, and never quiet about it. Half a scaffold does not run: the first # chapter imports the claim vocabulary, so a missing `status.py` is a collection error in # somebody else's project with nothing pointing at the cause. raise ScaffoldUnavailable( f"this build of Plumb is not carrying {', '.join(missing)}, so the spec it would " "write could not run. Reinstall the tool, or run from a checkout") written, kept = [], [] for name, text in files.items(): written, kept = _place(spec / name, text, root, written, kept) return _place(spec / FLOW_DIR / FIRST_FLOW, FLOW.format(module=FIRST_MODULE, ns=namespace), root, written, kept) def _place(target: Path, text: str, root: Path, written: list, kept: list): name = str(target.relative_to(root)) if target.exists(): kept.append(name) else: target.parent.mkdir(parents=True, exist_ok=True) target.write_text(text) written.append(name) return written, kept PKi\H plumb/store.py"""The contributor's own store, and the outbox behind it (``ch6-6``, ``ch6-7``, ``ch6-9``). A record is one file, which is what makes withdrawal an operation rather than an excavation. The outbox is a directory of marker files rather than an in-memory list, because ``ch6-7``'s promise that an unreachable endpoint costs the contributor nothing is empty if the queue dies with the process. """ import re from datetime import datetime, timezone from pathlib import Path from plumb.friction import FrictionRecord, load_all MAX_ID_LENGTH = 72 class FrictionStore: """Records under ``root``; anything awaiting transmission marked in ``root/outbox``.""" def __init__(self, root: Path | str): self.root = Path(root) self.outbox = self.root / "outbox" def _ensure(self) -> None: self.outbox.mkdir(parents=True, exist_ok=True) def file(self, record: FrictionRecord, *, queue: bool = False) -> Path: """Write a record, optionally marking it to be sent. Validation already happened at construction (``ch6-4``), so nothing invalid can reach the disk.""" self._ensure() path = record.write_to(self.root) if queue: self.enqueue(record.id) return path def all(self) -> list[FrictionRecord]: return load_all(self.root) if self.root.is_dir() else [] def get(self, record_id: str) -> FrictionRecord: path = self.root / f"{record_id}.json" if not path.is_file(): raise KeyError(record_id) return FrictionRecord.from_json(path.read_text()) def withdraw(self, record_id: str) -> bool: """Remove the record and anything pending for it (``ch6-9``). Idempotent: withdrawing what is already gone is not an error, because a contributor should never have to know whether their first attempt landed.""" removed = False path = self.root / f"{record_id}.json" if path.is_file(): path.unlink() removed = True self.dequeue(record_id) return removed # ── the outbox (ch6-7) ──────────────────────────────────────────────────── def enqueue(self, record_id: str) -> None: self._ensure() (self.outbox / record_id).touch() def dequeue(self, record_id: str) -> None: marker = self.outbox / record_id if marker.exists(): marker.unlink() def queued(self) -> list[str]: if not self.outbox.is_dir(): return [] return sorted(p.name for p in self.outbox.iterdir() if p.is_file()) def mint_id(observation: str, *, now: datetime | None = None) -> str: """A stable, human-meaningful id fixed at capture (``ch6-11``). Date plus a slug of the observation's opening sentence: readable in a directory listing, sortable by when it was seen, and derived from the record rather than from a counter that would need somewhere to live. """ stamp = (now or datetime.now(timezone.utc)).date().isoformat() opening = re.split(r"(?<=[.!?])\s", observation.strip(), maxsplit=1)[0] slug = re.sub(r"[^a-z0-9]+", "-", opening.lower().replace("'", "")).strip("-") if not slug: raise ValueError("cannot mint an id from an observation with no words in it") return f"{stamp}-{slug}"[:MAX_ID_LENGTH].rstrip("-") def utc_now() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat() PK\plumb/templates/PK|\09  plumb/templates/status.py"""The gold-spec's claim vocabulary — deliberately minimal. A story authors nothing but its own existence. Status is **derived, never written**: Plumb computes "proven" from a test that cites the story (a ``proves`` marker) and, when that test ran, passed, and its execution matched the kind of claim the story makes — a behavioral story grounding into production code, a structural one staying inert. Mutation is **not** part of that gate: it is a free opt-in beside it (``ch0-3``, ``ch4-2``), and a story proves with it switched off. Priority, estimates and required depth — the grooming overlay — are absent here for exactly one reason: none of them is a claim, and this vocabulary carries claims. That is a statement about *this file*, not about the product. Grooming needs no AI, so it is free (``ch7-2``); it is unprovable, so it sits outside the **gate** rather than outside the tool. Three separable exclusions — from the vocabulary, from the gate, and from the free tier — and only the first two apply. There is intentionally no ``gap()`` / ``mvp()`` / ``realized_by()``. Those were *authored* status; here status is not authored at all. """ import pytest try: from plumb.adapters.python.archetypes import declare except ImportError: # pragma: no cover - exercised by a scaffolded project, not by this one # A spec must run under the project's OWN test runner, with or without Plumb importable # (`ch9-1`). Plumb installs as a self-contained artifact, not as a library in every # consumer's environment, so plain `pytest` has no `plumb` module — and a spec that only runs # under the tool is a spec nobody can run while writing it. # # The archetype is the only thing lost, and losing it is safe: behavioral is the documented # default (`ch1-7`), so an undeclared archetype is the same answer a spec that never mentions # one gives. Degrade and surface rather than fail (`ch0-7`); `plumb board` supplies it. def declare(story_id: str, archetype: str = "behavioral") -> None: return None def story(id: str, archetype: str = "behavioral") -> None: """Declare a requirement node — the whole authored vocabulary. Status is Plumb's to derive from a citing test, not stated here. Until Plumb verifies such a test (ran + passed + the archetype's check), the story reads as UNPROVEN — so a fresh spec with no implementation is all-unproven and green, because skipped is not failed. ``archetype`` says what *kind* of claim this is, and so what a real proof of it looks like: **behavioral** is proved by running the wired system, **structural** by inspecting its shape (``ch1-7``). It is not a status and cannot become one — it selects which check the gate applies, never whether that check passes, and a story declared structural can never be proved by executing anything. Behavioral is the default, so a spec that never mentions archetypes is unchanged. """ declare(id, archetype) pytest.skip(f"UNPROVEN[{id}] — proven when a citing test passes Plumb's gates") PK\\$plumb/templates/test_spec_backlog.py"""The gold-spec backlog, printed by ``pytest`` — no external tooling. Story ids come from the chapter modules (functions named ``chN_M``); each story's one-line DESCRIPTION is joined from its markdown trace table. Status is derived, so today every story is UNPROVEN — the backlog is the whole ideal, awaiting the tests that will prove it. Status and prose each have one home; this only reads and joins them into a single ``pytest`` run. """ import re from pathlib import Path from conventions import FLOW_DIR, MODULE_GLOB, NAMESPACE SPEC_DIR = Path(__file__).parent DIAGRAMS = SPEC_DIR / FLOW_DIR # Story functions are N_M, and nested sub-stories N_M_K (a subroutine's internals). STORY = re.compile(rf"def\s+({NAMESPACE}\d+(?:_\d+)+)\s*\(") TRACE_ROW = re.compile(rf"^\|\s*`({NAMESPACE}\d+(?:-\d+)+)`\s*\|\s*(.+?)\s*\|", re.MULTILINE) def chapter_files() -> list[Path]: return sorted(SPEC_DIR.glob(MODULE_GLOB)) def stories() -> list[str]: found = [] for path in chapter_files(): for m in STORY.finditer(path.read_text()): found.append(m.group(1).replace("_", "-")) return found def descriptions() -> dict[str, str]: found: dict[str, str] = {} if not DIAGRAMS.is_dir(): return found for md in sorted(DIAGRAMS.glob("*.md")): for story_id, text in TRACE_ROW.findall(md.read_text()): found.setdefault(story_id, text.replace("**", "").strip()) return found def test_print_backlog(capsys): desc = descriptions() found = stories() out = [ "", "══════ GOLD-SPEC BACKLOG ══════", f"{len(found)} stories · all UNPROVEN until a citing test passes Plumb's gates", "", ] out += [f" · {s} — {desc.get(s, '(no trace-table description)')}" for s in found] out.append("═══════════════════════════════") with capsys.disabled(): print("\n".join(out)) assert found, "expected story functions (chN_M) in the chapter modules" PK\sp##-plumb/templates/test_spec_status_guardrail.py"""Guardrails that keep the spec honest without running any implementation. Pure source scan — no build, no network — so it always runs. It checks the spec's own consistency: every story declares itself via ``story(...)`` and nothing else (status is derived, never authored), every story carries a one-line description, every story is placed in its chapter's process flow (IPO), no implementation test cites a story the spec does not define, and every open unknown is uniquely named and tracked in both of its two homes. It does NOT judge whether a story is *proven* — that is Plumb's job, and it needs the real tests to do it. """ import re from pathlib import Path from conventions import CITATION_MARKER, FLOW_GLOB, NAMESPACE from test_spec_backlog import DIAGRAMS, STORY, chapter_files, descriptions, stories REPO = Path(__file__).parent.parent CITATION = re.compile(rf"""mark\.{CITATION_MARKER}\(\s*["']({NAMESPACE}[\w-]+)["']""") UNKNOWN = re.compile(rf"\b{NAMESPACE}\d+(?:-\d+)*-U\d+\b") UNKNOWN_BULLET = re.compile(r"^- \*\*(\S+) —", re.M) def diagram_files() -> list[Path]: return sorted(DIAGRAMS.glob(FLOW_GLOB)) def chapter_prefix(md: Path) -> str: """``chapter-01.2-gate.md`` → ``ch1-2`` — the namespace every id authored in that file carries.""" stem = FLOW_GLOB.replace("*", "") major, minor = re.match(rf"{re.escape(stem.rstrip('.md').rstrip('-'))}-(\d+)(?:\.(\d+))?", md.name).groups() return f"{NAMESPACE}{int(major)}" + (f"-{int(minor)}" if minor else "") def declared_unknowns(md: Path) -> list[str]: """The ids the chapter's Open-unknowns bullets introduce. That section is where an unknown is *declared*; the diagram node then has to match it.""" text = md.read_text() if "## Open unknowns" not in text: return [] return UNKNOWN_BULLET.findall(text.split("## Open unknowns")[1].split("\n## ")[0]) def _story_bodies(): for path in chapter_files(): src = path.read_text() matches = list(STORY.finditer(src)) for i, m in enumerate(matches): end = matches[i + 1].start() if i + 1 < len(matches) else len(src) yield path, m.group(1), src[m.end():end] def test_every_story_declares_itself_via_story_only(): violations = [f"{p.name}#{name}" for p, name, body in _story_bodies() if "story(" not in body] assert not violations, ( "a story must declare itself via story(...) — status is derived, never authored, so there is " f"no gap()/mvp()/realized_by() to write: {violations}" ) def test_every_story_has_a_markdown_description(): described = descriptions() missing = [s for s in stories() if s not in described] assert not missing, ( "every story needs a one-line row in its chapter's Story → test trace table " f"(| `chN-M` | what it proves |) — an undescribed story is an un-triaged to-do: {missing}" ) def test_every_story_appears_in_its_chapter_ipo(): """A story described but absent from its chapter's Input → process → output is a story nobody has placed in the flow. The flow is the spine; a story off it is not yet a requirement.""" missing = [] for md in diagram_files(): text = md.read_text() ch = chapter_prefix(md) rows = set(re.findall(rf"^\| `({ch}-\d+(?:-\d+)*)`", text, re.M)) if "## Input → process → output" not in text: missing.append(f"{md.name}: no IPO section") continue ipo = text.split("## Input → process → output")[1].split("\n## ")[0] placed = set(re.findall(rf"`({ch}-\d+(?:-\d+)*)`", ipo)) missing += [f"{md.name}#{s}" for s in sorted(rows - placed)] assert not missing, ( "every story must appear in its chapter's Input → process → output, or it is described but " f"not placed in the flow: {missing}" ) def test_no_implementation_test_cites_an_undefined_story(): """The citation check is vacuous until the first implementation test lands, and live from the moment it does — a dangling citation is a red run, not silent rot.""" defined = set(stories()) dangling = [] tests_dir = REPO / "tests" if tests_dir.is_dir(): for path in sorted(tests_dir.glob("**/*.py")): for cited in CITATION.findall(path.read_text()): if cited not in defined: dangling.append(f"{path.relative_to(REPO)} cites {cited}") assert not dangling, ( "an implementation test cites a story the gold-spec does not define; renumbering a story is " f"mechanical, an orphaned citation is a red run: {dangling}" ) def test_every_open_unknown_is_in_both_the_diagram_and_the_text(): """An open unknown is tracked in two places on purpose — a node in the chapter's diagram (attached to the story it influences, so a blocker is visible in the flow) AND a description in the chapter's Open-unknowns section. This gate keeps the two from drifting silently: the sets must match exactly, so the "two places" cost nothing.""" problems = [] for md in diagram_files(): mermaid = "\n".join(re.findall(r"```mermaid(.*?)```", md.read_text(), re.S)) own = f"{chapter_prefix(md)}-U" in_diagram = {u for u in UNKNOWN.findall(mermaid) if u.startswith(own)} in_text = set(declared_unknowns(md)) if in_diagram != in_text: problems.append( f"{md.name}: diagram {sorted(in_diagram) or '[]'} vs text {sorted(in_text) or '[]'}" ) assert not problems, ( "every open unknown must appear BOTH as a node in the chapter's diagram (attached to the story " f"it influences) AND as a description in its Open-unknowns section — these disagree: {problems}" ) def test_no_flow_edge_points_at_a_node_that_does_not_exist(): """A flow's completeness is mechanically checkable (`ch7-9`), and this is its cheapest half: an edge to an undefined node renders as an empty box and reads as a step nobody named. Written because retiring one open unknown left `CITE -.exposes.-> U5` behind and every other gate here passed — the unknown checks compare `chN-U` *ids*, and a mermaid node *name* is not one. A gate that cannot see a dangling edge lets the diagram rot while reporting the spec honest. """ # A node is declared wherever its name is followed by a bracket, which in mermaid may be at # the start of a line or inline in the middle of an edge — `D1 -->|no| D3{...}` does both. declares = re.compile(r"(\w+)\s*[\[({>]") arrows = re.compile(r"-{2,3}>|-\.-*>|-\.-|={2,}>") dangling = [] for md in diagram_files(): mermaid = "\n".join(re.findall(r"```mermaid(.*?)```", md.read_text(), re.S)) nodes = set(declares.findall(mermaid)) for line in mermaid.splitlines(): stripped = line.strip() if stripped.startswith(("subgraph", "classDef", "class ", "%%")): continue if not arrows.search(stripped): continue for part in arrows.split(re.sub(r"\|[^|]*\|", " ", stripped)): name = part.strip().split(":::")[0].strip() # A bare identifier — anything bracketed declared itself right here. if re.fullmatch(r"\w+", name) and name not in nodes: dangling.append(f"{md.name}: {name}") assert not dangling, ( f"a flow edge names a node the flow does not define: {sorted(set(dangling))}" ) def test_every_open_unknown_id_carries_its_chapter(): """Unknowns are cited by id across chapters — ``ch3-6`` leans on ``ch1-2-U2``, chapter 4 is the mechanical dent in ``ch1-2-U1``. A bare ``U2`` would name something different in every file, so the chapter is part of the id, exactly as it is for a story.""" wrong = [] for md in diagram_files(): prefix = chapter_prefix(md) wrong += [ f"{md.name}#{u}" for u in declared_unknowns(md) if not re.fullmatch(rf"{prefix}-U\d+", u) ] assert not wrong, ( "an open unknown's id must be its chapter's namespace plus U (ch3-U1, ch1-2-U2) so it reads " f"the same from any chapter — these are bare or misfiled: {wrong}" ) def test_no_open_unknown_id_is_declared_twice(): """Chapter-scoped ids are unique by construction only while one chapter owns one namespace. This is the guard that keeps that true, so a cross-chapter citation can never be ambiguous.""" owner: dict[str, str] = {} clashes = [] for md in diagram_files(): for u in declared_unknowns(md): if u in owner: clashes.append(f"{u}: {owner[u]} and {md.name}") owner[u] = md.name assert not clashes, ( f"an open unknown id must be declared in exactly one place — these are declared twice: {clashes}" ) PK\LFplumb/transmit.py"""Sending a record, and not caring much whether it arrives (``ch6-6``, ``ch6-7``). Capture never blocks the work. An unreachable endpoint is not an error the contributor should ever see: the record is already safe on their disk, it is marked in the outbox, and the next attempt drains it. Nothing here raises into a caller's workflow — the return value says what happened and the work carries on either way. """ import json import urllib.error import urllib.request from dataclasses import dataclass from pathlib import Path from plumb.enrolment import Enrolment, NotEnrolled, load from plumb.friction import FrictionRecord from plumb.store import FrictionStore TIMEOUT_SECONDS = 10 #: The record shape this client speaks, announced on every request (``ch6-13``). PROTOCOL_VERSION = 1 VERSION_HEADER = "Plumb-Record-Version" #: What an attempt came to. Three outcomes rather than two, because "it arrived" and "stop #: trying" are different facts and collapsing them is how a refused record got reported as sent. ACCEPTED, RETRY, REFUSED = "accepted", "retry", "refused" #: The receiver positively identified the *content* as unusable, and no retry changes that. #: Everything else non-2xx is retried: a version the receiver does not know yet becomes #: deliverable when it is upgraded, and discarding is irreversible where a retry costs a request #: (``ch6-13``). Until a receiver speaks the versioned protocol, drift arrives as a bare 400 and #: is indistinguishable from bad content — so 400 is retried, and 422 is what a versioned #: receiver uses to say the content itself is wrong. TERMINAL = frozenset({413, 422}) @dataclass(frozen=True) class Outcome: """What became of an attempt. Never an exception — see the module docstring.""" sent: tuple[str, ...] = () queued: tuple[str, ...] = () refused: tuple[str, ...] = () reason: str = "" @property def ok(self) -> bool: return not self.queued and not self.refused and not self.reason def deliver(record: FrictionRecord, enrolment: Enrolment) -> tuple[bool, str]: """One attempt. Returns ``(verdict, detail)`` — never raises for a network condition. The verdict is one of :data:`ACCEPTED`, :data:`RETRY` or :data:`REFUSED`. It was a boolean once, where ``True`` meant "stop trying" and was read as "arrived", so a refused record was dequeued and reported as sent (``ch6-13``). """ request = urllib.request.Request( url=enrolment.endpoint.rstrip("/") + "/friction", data=record.to_json().encode("utf-8"), method="POST", headers={ "Content-Type": "application/json", "Authorization": f"Bearer {enrolment.credential}", VERSION_HEADER: str(PROTOCOL_VERSION), }, ) try: with urllib.request.urlopen(request, timeout=TIMEOUT_SECONDS) as response: if response.status in (200, 201, 409): # 409 = already held; it did arrive return ACCEPTED, f"HTTP {response.status}" return RETRY, f"HTTP {response.status}" except urllib.error.HTTPError as refused: if refused.code == 409: return ACCEPTED, "HTTP 409 — already held" if refused.code == 426: # The receiver named a version mismatch. Temporary by nature: it becomes deliverable # the moment that receiver is upgraded (`ch6-13`). return RETRY, f"HTTP 426 — {_accepts(refused)}, will retry" if refused.code in TERMINAL: return REFUSED, f"HTTP {refused.code} — refused, not retried" return RETRY, f"HTTP {refused.code} — will retry" except (urllib.error.URLError, OSError, TimeoutError) as unreachable: return RETRY, f"unreachable: {unreachable}" def _accepts(refused: urllib.error.HTTPError) -> str: """What the other end said it speaks. A refusal that names nothing is a dead end.""" try: accepts = json.loads(refused.read().decode("utf-8")).get("accepts") return f"receiver accepts {accepts}, this client speaks {PROTOCOL_VERSION}" except Exception: return f"version mismatch; this client speaks {PROTOCOL_VERSION}" def send_one(record: FrictionRecord, store: FrictionStore, enrolment_path: Path | None = None) -> Outcome: """File-and-forget: deliver if enrolled and reachable, otherwise leave it queued.""" try: enrolment = load(enrolment_path) except (NotEnrolled, ValueError, json.JSONDecodeError) as not_enrolled: return Outcome(reason=str(not_enrolled)) # local by default (ch6-6) — not a failure verdict, detail = deliver(record, enrolment) if verdict == ACCEPTED: store.dequeue(record.id) return Outcome(sent=(record.id,), reason=detail) # Refused or retryable, the record stays in the outbox either way. A refusal is the record a # human should look at, and quietly clearing it from the outbox is how it gets forgotten — # removal is an operation somebody performs (`ch6-9`), never a side effect of a rejection. store.enqueue(record.id) if verdict == REFUSED: return Outcome(refused=(record.id,), reason=detail) return Outcome(queued=(record.id,), reason=detail) def drain(store: FrictionStore, enrolment_path: Path | None = None) -> Outcome: """Retry everything in the outbox. Stops at the first unreachable attempt rather than hammering a down endpoint once per queued record.""" try: enrolment = load(enrolment_path) except (NotEnrolled, ValueError, json.JSONDecodeError) as not_enrolled: return Outcome(reason=str(not_enrolled)) sent, still_queued, refused, reason = [], [], [], "" for record_id in store.queued(): if reason: still_queued.append(record_id) continue try: record = store.get(record_id) except KeyError: store.dequeue(record_id) # withdrawn while queued (ch6-9); nothing to send continue verdict, detail = deliver(record, enrolment) if verdict == ACCEPTED: store.dequeue(record_id) sent.append(record_id) elif verdict == REFUSED: # Refused on content: keep it, name it, and carry on draining the rest. One bad # record must not stall the queue behind it, and must not vanish from it either. refused.append(record_id) else: still_queued.append(record_id) reason = detail # a retryable condition stops the drain; the rest stay queued return Outcome(sent=tuple(sent), queued=tuple(still_queued), refused=tuple(refused), reason=reason) PK{\4AT plumb/triggers.py"""The trigger conditions that ship with the tool (``ch6-10``). An agent files nothing it has to go looking for, so the instruction telling it *when* to file travels with the capability rather than living in a wiki nobody opens. These are concrete conditions, not an exhortation to be observant: each one is a thing that either happened in the last few minutes or did not. Emitted by ``plumb friction why`` so an agent can read it without leaving the tool. """ TRIGGERS = ( ("re-read", "A spec section had to be re-read to be understood. The first reading failing is " "the observation — not that the text is bad, but that it did not land."), ("abandoned", "An attempt was started, abandoned, and retried differently. The abandoned path " "is the record; nobody remembers it an hour later."), ("interpreted", "A board message, error or condition code needed interpreting before it could " "be acted on. If it needed a translator, it is not yet an interface."), ("worked-around", "Something was done a longer way because the short way did not exist. This " "is the one that most reliably goes unfiled, because it feels like progress."), ("surprised", "Behaviour differed from what the spec led you to expect — either is a defect: " "the code, or the spec that set the expectation."), ("asked", "A question had to be asked of a person because no artifact answered it. The gap is " "in the artifact, and it closes the moment the answer is given and lost."), ) BAR = ( "A record must say what the rub implies FOR THE METHOD. If you cannot state the " "implication, do not file: an observation without one is an anecdote, and a tireless filer " "buries the signal for everyone. The bar is deliberate, and it is the same shape as the proof " "gate — no claim without the thing that makes it real." ) def instruction() -> str: """The shipped instruction, as an agent should read it.""" lines = [ "FILE A FRICTION RECORD WHEN — any of these, at the moment it happens:", "", ] lines += [f" {name:<14} {text}" for name, text in TRIGGERS] lines += [ "", "THE BAR", "", f" {BAR}", "", "HOW", "", " plumb friction file --observation '...' --implication '...' [--story APP-142]", " [--code PATH:START-END] [--tag NAME]", "", "Capture is local by default and never transmits unless this installation was enrolled", "Filing writes nothing to the project you are working in.", ] return "\n".join(lines) PK\-1wh55 __main__.pyPK\Auplumb/PK\3s00plumb/__init__.pyPK\Aplumb/adapters/PK=\'ZZ%plumb/adapters/__init__.pyPK\Aplumb/adapters/java/PK\WԄnn"plumb/adapters/java/plumb-java.jarPK\Aplumb/adapters/python/PK=\INC##!˔plumb/adapters/python/__init__.pyPK \W !-plumb/adapters/python/__main__.pyPK \t,, Yplumb/adapters/python/adapter.pyPK=\1X::#)plumb/adapters/python/archetypes.pyPK\Z*44"plumb/adapters/python/grounding.pyPK{\^RR plumb/adapters/python/mutants.pyPK3\a::!plumb/adapters/python/mutation.pyPK3\#S =plumb/adapters/python/surface.pyPK{\ʜ**Pplumb/adapters/python/trace.pyPKG\GG {plumb/cli.pyPK \f!plumb/config.pyPK{\ ADplumb/core/PK=\]ccmplumb/core/__init__.pyPK{\*#Fplumb/core/board.pyPK\(\w0w0plumb/core/gate.pyPK{\R;;d)plumb/core/manifest.pyPK}\& & eplumb/enrolment.pyPKN}\ݕJJtpplumb/friction.pyPK\442?2?plumb/launch.pyPK \tzY#Y#Lplumb/receiver.pyPK \o0plumb/registry_ops.pyPK \3  plumb/runfile.pyPK\QN+p.p. plumb/scaffold.pyPKG\b7plumb/spec_scaffold.pyPKi\H kQplumb/store.pyPK\Aq_plumb/templates/PK|\09  _plumb/templates/status.pyPK\\$kplumb/templates/test_spec_backlog.pyPK\sp##-Ntplumb/templates/test_spec_status_guardrail.pyPK\LFplumb/transmit.pyPK{\4AT plumb/triggers.pyPK''o