feat(proactive): prepare controlled private catalogue delivery and reject repeated comparisons

app/Services/ProactiveSupply.php

@@ -73,9 +73,12 @@ final class ProactiveSupply
         if(!$v)return [];
         abort_unless(Schema::hasTable('proactive_supply_observations'),422,'Base interna indisponível.');
         return $this->read($actor,function()use($v){
+            $fingerprints=[];
             foreach($v as $row){
                 $o=DB::table('proactive_supply_observations')->where('id',$row['observation_id'])->first();
                 if(!$o)throw \Illuminate\Validation\ValidationException::withMessages(['proactive.supply_comparisons'=>'Registo da oferta indisponível; preserve a referência anterior e confira o ambiente.']);
+                if($o->duplicate_of!==null||isset($fingerprints[$o->fingerprint]))throw \Illuminate\Validation\ValidationException::withMessages(['proactive.supply_comparisons'=>'Esta observação administrativa é repetida. Compare a referência original uma única vez; a repetição e a sua fonte permanecem conservadas na base interna.']);
+                $fingerprints[$o->fingerprint]=true;
             }
             return $v;
         });

docs/investmoz/unification/proactive-supply-delivery-runbook-20261028.md

@@ -0,0 +1,28 @@
+# Entrega controlada da base comparativa interna
+
+A entrega inclui o candidato QA `a022ed98`. As sete provas dirigidas/62 asserções, tipos, compilação e percurso de Gurué continuam aceites. O delta de entrega acrescenta o carregador técnico e impede que a API grave uma repetição preservada como outra comparação; este último delta tem uma prova dirigida própria, sem repetir a suite anterior.
+
+361.464 é o número de observações administrativas, incluindo 147.165 repetições preservadas. As 214.299 projecções distintas também **não** são empresas únicas. A consulta exclui `duplicate_of` das contagens e dos resultados; a gravação recusa repetições como comparações adicionais. A cobertura nacional é parcial. Ausência de resultados significa informação insuficiente. A comparação fundamenta uma hipótese e não confirma procura, capacidade actual, contrato ou viabilidade.
+
+## Sequência da entrega
+
+1. Conferir `main`, a revisão servida, ausência de deriva do código e capacidade. A base da entrega é `569a1cd7`; não usar force-push nem substituir alterações posteriores.
+2. Criar backup novo de base e ficheiros/configuração. Preservar os backups anteriores. Restaurar numa instância isolada, sem rede e na porta55441, e comparar as174 tabelas públicas do inventário.
+3. Ensaiar o SQL da migração exacta, o carregamento e a sua repetição. Conferir que as174 tabelas anteriores não mudam. Verificar a conta real da aplicação: sem contexto de actor não lê a base; com o contexto autorizado recupera435 observações distintas em Gurué e conserva237 repetições fora dessa contagem. Não criar utilizadores, organizações ou dossiers fictícios no activo.
+4. Conferir também backup/restauro das duas novas tabelas, incluindo RLS forçada. Registar espaço efectivamente ocupado pelo ensaio e voltar a conferir capacidade antes da instalação.
+5. Na sessão SSH autenticada do processo existente, aplicar a migração delimitada e registar uma linha de migração. Conceder apenas leitura nas duas tabelas à conta de aplicação já existente. Carregar o pacote privado em transacção, com hashes e contagens fixos, recusa de conflitos e sem sobrescrever observações. Não publicar listas nem criar anúncios, clientes confirmados ou promotores.
+6. Acrescentar apenas o hash dessa migração ao contrato de schema do instalador. Conservar os90 hashes anteriores e o próprio programa do instalador. Este passo é necessário porque o processo Git não executa migrações ou importações de conteúdo automaticamente.
+7. Fazer a entrega normal por Git/`main`, mantendo a construção isolada e as guardas habituais do instalador. As provas funcionais aceites são reutilizadas; a CI habitual do instalador permanece.
+8. Conferir revisão pública, hashes do código, schema e permissões esperados, ficheiros/configuração e linhas anteriores, separando a nova linha de migração e o delta operacional de cache/sessões. Conferir negação de acesso anónimo e consulta autorizada de Gurué. A conferência visual autenticada no portal é distinta do inventário técnico por SSH.
+
+## Carregador
+
+`tools/load-proactive-supply.py` usa pré-visualização por omissão. Exige o pacote privado com `import-manifest.json`, `preview.json`, `scan.json`, `schema.sql` e `observations.jsonl.gz`. O JSONL aceite tem SHA256 `dec9db62a52a8bac5f1e2675aecbeb4239f2b1f4fa521db01693c32bdfeef01e`. Os quatro originais e as datas são os aceites em QA; não se pede nova validação genérica da sua origem.
+
+O modo `rehearsal` exige container sem rede/portas publicadas e a identidade efectiva `proactive_supply_delivery_20261028_test:55441`. O modo `apply` exige a sessão técnica SSH, a identidade `teste_investidor:5432`, revisão esperada e backup/restauro verificado. A preparação e os recibos ficam privados em `/var/backups/investmoz-proactive-supply-delivery-20261028`, fora da área servida. O carregador não cria schema no activo, não publica dados e não apaga ou substitui registos. A tabela temporária é indexada antes de conferir a linhagem de duplicações.
+
+## Preservação e recuo
+
+Se uma conferência divergir, preservar os exemplares e delimitar a diferença antes de agir. Recuar código não autoriza apagar a base privada ou o seu histórico. A migração recusa `down` enquanto existirem fontes carregadas. Um restauro integral nunca deve sobrescrever intervenções posteriores sem reconciliação; os originais e o pacote privado permitem recuperar o catálogo separadamente. As duas novas tabelas e o único registo de migração são o delta de schema/dados previsto; não alterar papéis de funcionários ou permissões anteriores.
+
+As restrições de análise interna, preservação, não duplicação, não publicação e separação QA/activo vêm das decisões expressas da Tempo e do `AGENTS.md`. Identidades exactas, hashes, limites técnicos, indexação temporária e contratos são escolhas de implementação, não novas exigências organizacionais. O incremento não fecha os requisitos nacionais amplos e não altera L8, que permanece parcial.

tests/Feature/ProactiveSupplyTest.php

@@ -84,4 +84,14 @@ final class ProactiveSupplyTest extends TestCase
             $this->assertNotNull(DB::table('proactive_supply_observations')->where('id',$this->observation)->first());
         }finally{DB::statement('RESET ROLE');}
     }
+    public function test_preserved_repetitions_cannot_inflate_saved_comparisons():void
+    {
+        $this->expectException(\Illuminate\Validation\ValidationException::class);
+        $this->expectExceptionMessage('Esta observação administrativa é repetida.');
+        $comparison=['relationship'=>'competition','rationale'=>'SYNTHETIC QA duplicate must not count as another competitor','limitations'=>'Historical observation,not a unique company'];
+        app(ProactiveSupply::class)->comparisons($this->actor,[
+            [...$comparison,'observation_id'=>$this->observation],
+            [...$comparison,'observation_id'=>hash('sha256','QA-REPEATED-OBSERVATION')],
+        ],null);
+    }
 }

tools/load-proactive-supply.py

@@ -0,0 +1,136 @@
+"""Controlled delivery of the accepted internal snapshot. Never publishes entities.
+
+Preview is the default. Apply is an authenticated deployment operation, guarded by
+the exact destination, accepted source hashes and a fresh verified backup/restore.
+The rehearsal uses a disposable network-isolated PostgreSQL container only.
+"""
+import argparse
+import datetime as dt
+import gzip
+import hashlib
+import json
+import os
+from pathlib import Path
+import subprocess
+
+BASELINE = '569a1cd750f3d2a0b87f130e992e73e070bbd1ab'
+RAW_SHA256 = 'dec9db62a52a8bac5f1e2675aecbeb4239f2b1f4fa521db01693c32bdfeef01e'
+ROWS = 361464
+FIELDS = ['id','source_hash','sheet','row_number','entity_key','name','province','district','cae','activity','sector','source_date','fingerprint','duplicate_of','identity_conflict','territory_id','territory_resolution']
+SOURCES = {
+    '1ba4c4114b16b407e7746712d00d4f681246bbbc1b42351e96a31f7129acad30':148556,
+    '1b5087b464e39e88c42def39655c2391aaaf684b4c832d6a77ff274b98889f00':212631,
+    'f02e086ed4973ef17234aa6cc54b98fbb7a6952ffb16ec4dc4f7d0c72af6400c':277,
+    '83b4f1200988e99834c2ae3915346fbf237d128308554b67b5a080d4956bf506':0,
+}
+
+def csv_value(value):
+    if value is None:return r'\N'
+    if isinstance(value,bool):value='t' if value else 'f'
+    value=str(value)
+    if any(c in value for c in ('\x00','\r','\n')):raise ValueError('Unexpected multiline/NUL source projection')
+    return '"'+value.replace('"','""')+'"'
+
+def fail_unless(ok,message):
+    if not ok:raise RuntimeError(message)
+
+def sql_command(container=None):
+    if container:return ['docker','exec','-i',container,'psql','-p','55441','-U','postgres','-d','proactive_supply_delivery_20261028_test','-XqAt','-v','ON_ERROR_STOP=1']
+    return ['runuser','-u','postgres','--','psql','-d','teste_investidor','-XqAt','-v','ON_ERROR_STOP=1']
+
+def query(command,text):
+    return subprocess.check_output(command,input=text.encode(),stderr=subprocess.PIPE).decode().strip()
+
+def prepare(package,output):
+    manifest=json.loads((package/'import-manifest.json').read_text())
+    preview=json.loads((package/'preview.json').read_text())
+    scan=json.loads((package/'scan.json').read_text())
+    schema=package/'schema.sql'
+    fail_unless(hashlib.file_digest(schema.open('rb'),'sha256').hexdigest()==manifest['schema_sql_sha256'],'Schema SQL differs from reviewed candidate migration')
+    fail_unless(manifest['candidate']=='a022ed9848b41ba342b4209414c2bbf2ebd0a7a3','Unexpected candidate')
+    fail_unless(set(s['sha256']for s in preview['sources'])==set(SOURCES),'Unexpected originals')
+    fail_unless(scan['exit_code']==0 and set(scan['source_hashes'])==set(SOURCES),'Hash-specific source scan receipt missing')
+    fail_unless(preview['records']==ROWS,'Incomplete preview')
+    fail_unless(hashlib.file_digest((package/'observations.jsonl.gz').open('rb'),'sha256').hexdigest()==manifest['compressed_sha256'],'Compressed snapshot changed')
+    rawhash=hashlib.sha256();counts=dict.fromkeys(SOURCES,0);n=0
+    seen=set()
+    with output.open('w',encoding='utf-8',newline='\n')as sql:
+        os.chmod(output,0o600)
+        sql.write("BEGIN;\nSET LOCAL lock_timeout='10s';\nSELECT pg_advisory_xact_lock(hashtext('proactive-supply-delivery-v1'));\n")
+        sql.write("CREATE TEMP TABLE supply_delivery_sources (LIKE public.proactive_supply_sources INCLUDING DEFAULTS) ON COMMIT DROP;\n")
+        sql.write("CREATE TEMP TABLE supply_delivery_observations (LIKE public.proactive_supply_observations INCLUDING DEFAULTS) ON COMMIT DROP;\n")
+        sql.write("COPY supply_delivery_sources (sha256,name,period,limitations,receipt,imported_at) FROM STDIN WITH (FORMAT csv,NULL '\\N');\n")
+        now=dt.datetime.now(dt.timezone.utc).isoformat()
+        for s in preview['sources']:
+            receipt={'original':s,'transport_sha256':RAW_SHA256,'candidate':manifest['candidate'],'scan':scan,'access':'internal','publication':False,'counting':'administrative observations;not unique companies;repeated projections retained but excluded from query total'}
+            sql.write(','.join(map(csv_value,[s['sha256'],s['name'],s['period'],preview['limitations'],json.dumps(receipt,ensure_ascii=False),now]))+'\n')
+        sql.write('\\.\n')
+        sql.write("COPY supply_delivery_observations ("+','.join(FIELDS)+") FROM STDIN WITH (FORMAT csv,NULL '\\N');\n")
+        with gzip.open(package/'observations.jsonl.gz','rb')as stream:
+            for line in stream:
+                rawhash.update(line);r=json.loads(line)
+                fail_unless(set(r)==set(FIELDS),'Unexpected observation fields')
+                fail_unless(r['source_hash']in SOURCES,'Unknown source')
+                expected=hashlib.sha256((r['source_hash']+'|'+r['sheet']+'|'+str(r['row_number'])).encode()).hexdigest()
+                fail_unless(r['id']==expected and r['id']not in seen,'Invalid or repeated row identity')
+                seen.add(r['id']);n+=1;counts[r['source_hash']]+=1
+                sql.write(','.join(csv_value(r[k])for k in FIELDS)+'\n')
+        sql.write('\\.\n')
+        fail_unless(rawhash.hexdigest()==RAW_SHA256 and n==ROWS and counts==SOURCES,'Snapshot checksum/count mismatch')
+        sql.write(r'''
+CREATE UNIQUE INDEX supply_delivery_observations_id ON supply_delivery_observations(id);
+ANALYZE supply_delivery_observations;
+DO $$ BEGIN
+ IF (SELECT count(*) FROM supply_delivery_observations)<>361464 THEN RAISE EXCEPTION 'Incomplete internal base'; END IF;
+ IF (SELECT count(*) FROM supply_delivery_observations WHERE duplicate_of IS NOT NULL)<>147165 THEN RAISE EXCEPTION 'Repeated projection count differs'; END IF;
+ IF EXISTS(SELECT 1 FROM supply_delivery_observations x LEFT JOIN supply_delivery_observations y ON y.id=x.duplicate_of WHERE x.duplicate_of IS NOT NULL AND (y.id IS NULL OR x.id=y.id OR x.fingerprint<>y.fingerprint)) THEN RAISE EXCEPTION 'Invalid duplicate lineage'; END IF;
+ IF EXISTS(SELECT 1 FROM supply_delivery_observations x LEFT JOIN public.geo_locations g ON g.id=x.territory_id WHERE x.territory_id IS NOT NULL AND (g.id IS NULL OR g.deleted_at IS NOT NULL)) THEN RAISE EXCEPTION 'Destination geography differs'; END IF;
+ IF EXISTS(SELECT 1 FROM supply_delivery_observations x JOIN public.proactive_supply_observations y ON x.id=y.id WHERE to_jsonb(x)<>to_jsonb(y)) THEN RAISE EXCEPTION 'Previously installed observation differs;no overwrite'; END IF;
+ IF EXISTS(SELECT 1 FROM supply_delivery_sources x JOIN public.proactive_supply_sources y ON x.sha256=y.sha256 WHERE (x.name,x.period,x.limitations) IS DISTINCT FROM (y.name,y.period,y.limitations)) THEN RAISE EXCEPTION 'Existing original metadata differs'; END IF;
+END $$;
+INSERT INTO public.proactive_supply_sources SELECT * FROM supply_delivery_sources ON CONFLICT(sha256) DO NOTHING;
+INSERT INTO public.proactive_supply_observations SELECT * FROM supply_delivery_observations ON CONFLICT(id) DO NOTHING;
+SELECT json_build_object('records',count(*),'distinct_observations',count(*) FILTER(WHERE duplicate_of IS NULL),'repeated_projections',count(*) FILTER(WHERE duplicate_of IS NOT NULL),'not_unique_companies',true,'published',false) FROM public.proactive_supply_observations;
+COMMIT;
+''')
+    return {'candidate':manifest['candidate'],'records':n,'source_counts':counts,'raw_sha256':rawhash.hexdigest(),'sql_bytes':output.stat().st_size,'mode':'prepared;no database write','unique_company_count':None,'duplicate_suppression':'WHERE duplicate_of IS NULL;retained administrative sightings do not count as separate offers','national_coverage':'partial','investment_conclusion':'hypothesis only;does not prove demand or viability'}
+
+def main():
+    p=argparse.ArgumentParser();p.add_argument('--package',type=Path,required=True);p.add_argument('--work',type=Path,required=True);p.add_argument('--mode',choices=['preview','rehearsal','apply'],default='preview');p.add_argument('--container');p.add_argument('--backup',type=Path)
+    a=p.parse_args();a.work.mkdir(mode=0o700,parents=True,exist_ok=True);output=a.work/'controlled-import.sql'
+    receipt=prepare(a.package,a.work/'controlled-import.sql')
+    if a.mode=='preview':print(json.dumps(receipt));return
+    fail_unless(os.geteuid()==0,'Authenticated technical deployment identity required')
+    if a.mode=='rehearsal':
+        fail_unless(a.container is not None,'Disposable QA container required')
+        info=json.loads(subprocess.check_output(['docker','inspect',a.container]))[0]
+        fail_unless(info['HostConfig']['NetworkMode']=='none' and not info['HostConfig'].get('PortBindings'),'QA must have no network or host ports')
+    else:
+        fail_unless(not a.container and os.environ.get('SSH_CONNECTION'),'Existing authenticated SSH deployment session required')
+        installed=json.loads(Path('/var/lib/investmoz-deploy/installed.json').read_text())
+        fail_unless(installed['commit']==BASELINE,'Reconcile installed code before preloading the new private base')
+        fail_unless(a.backup is not None,'Fresh backup/restore receipt required')
+        base=a.backup.resolve();fail_unless(base.is_relative_to('/var/backups/investmoz-proactive-supply-delivery-20261028'),'Unexpected backup root')
+        b=json.loads((base/'backup-receipt.json').read_text());r=json.loads((base/'restore-receipt.json').read_text())
+        stamp=dt.datetime.fromisoformat(b['created_at'].replace('Z','+00:00'))
+        fail_unless(dt.timedelta(0)<=dt.datetime.now(dt.timezone.utc)-stamp<=dt.timedelta(hours=24),'Backup is no longer fresh;take another before apply')
+        fail_unless(b['target']['database']=='teste_investidor' and int(b['target']['port'])==5432,'Backup identity mismatch')
+        fail_unless(r['restore_verified'] and r['source_dump_sha256']==b['sha256'],'Restore does not match backup')
+        fail_unless(hashlib.file_digest((base/'database.dump').open('rb'),'sha256').hexdigest()==b['sha256'],'Backup bytes changed')
+    command=sql_command(a.container)
+    identity=query(command,"SELECT current_database()||'|'||current_setting('port');")
+    expected='proactive_supply_delivery_20261028_test|55441'if a.mode=='rehearsal'else'teste_investidor|5432'
+    fail_unless(identity==expected,'Wrong actual database identity')
+    # Existing pipeline does not execute migrations. Schema installation is a
+    # separate, reviewed deployment step; this loader never creates it on active.
+    fail_unless(query(command,"SELECT count(*) FROM pg_class WHERE oid IN(to_regclass('public.proactive_supply_sources'),to_regclass('public.proactive_supply_observations')) AND relrowsecurity AND relforcerowsecurity;")=='2','Reviewed private schema/RLS not installed')
+    before=query(command,'SELECT count(*) FROM public.proactive_supply_observations;')
+    with output.open('rb')as stream:
+        result=subprocess.run(command,stdin=stream,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
+    fail_unless(result.returncode==0,'Import rolled back;'+result.stderr.decode(errors='replace')[-1000:])
+    lines=[line for line in result.stdout.decode().splitlines()if line.startswith('{')]
+    receipt.update(mode=a.mode,database=identity,before=int(before),after=json.loads(lines[-1]),idempotent_no_overwrite=True,active_existing_business_data_touched=False)
+    (a.work/'import-receipt.json').write_text(json.dumps(receipt,indent=2));os.chmod(a.work/'import-receipt.json',0o600)
+    print(json.dumps(receipt))
+
+if __name__=='__main__':main()