diff --git a/apps/api/app/main.py b/apps/api/app/main.py
index 83368ca..8c70d89 100644
--- a/apps/api/app/main.py
+++ b/apps/api/app/main.py
@@ -89,7 +89,7 @@ def _initialize_database(db_path: str) -> sqlite3.Connection:
         "pipeline_entries": (("notes", "TEXT NOT NULL DEFAULT ''"), ("next_action", "TEXT NOT NULL DEFAULT ''"), ("follow_up_at", "TEXT"), ("actor_user_id", "INTEGER"), ("idempotency_key", "TEXT"), ("version", "INTEGER NOT NULL DEFAULT 1")),
         "interactions": (("outcome", "TEXT NOT NULL DEFAULT 'other'"), ("notes", "TEXT NOT NULL DEFAULT ''"), ("next_action", "TEXT NOT NULL DEFAULT ''"), ("follow_up_at", "TEXT"), ("actor_user_id", "INTEGER"), ("idempotency_key", "TEXT")),
         "suppressions": (("active", "INTEGER NOT NULL DEFAULT 1"), ("updated_at", "TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP"), ("actor_user_id", "INTEGER")),
-        "sources": (("source_code", "TEXT NOT NULL DEFAULT ''"), ("display_name", "TEXT NOT NULL DEFAULT ''"), ("approved", "INTEGER NOT NULL DEFAULT 0"), ("policy_json", "TEXT NOT NULL DEFAULT '{}'"), ("quota_json", "TEXT NOT NULL DEFAULT '{}'")),
+        "sources": (("source_code", "TEXT NOT NULL DEFAULT ''"), ("display_name", "TEXT NOT NULL DEFAULT ''"), ("approved", "INTEGER NOT NULL DEFAULT 0"), ("policy_json", "TEXT NOT NULL DEFAULT '{}'"), ("quota_json", "TEXT NOT NULL DEFAULT '{}'"), ("owner", "TEXT NOT NULL DEFAULT ''"), ("terms_url", "TEXT NOT NULL DEFAULT ''"), ("terms_status", "TEXT NOT NULL DEFAULT 'unreviewed'"), ("rate_limit", "TEXT NOT NULL DEFAULT ''"), ("daily_quota", "INTEGER"), ("credentials_configured", "INTEGER NOT NULL DEFAULT 0")),
         "discovery_queries": (("selected_adapters_json", "TEXT NOT NULL DEFAULT '[]'"), ("location", "TEXT NOT NULL DEFAULT ''"), ("category", "TEXT NOT NULL DEFAULT ''"), ("max_records", "INTEGER NOT NULL DEFAULT 100"), ("daily_limit", "INTEGER NOT NULL DEFAULT 1000"), ("schedule", "TEXT NOT NULL DEFAULT ''"), ("dry_run", "INTEGER NOT NULL DEFAULT 0"), ("lifecycle", "TEXT NOT NULL DEFAULT 'draft'")),
         "discovery_runs": (("selected_adapters_json", "TEXT NOT NULL DEFAULT '[]'"), ("location", "TEXT NOT NULL DEFAULT ''"), ("category", "TEXT NOT NULL DEFAULT ''"), ("max_records", "INTEGER NOT NULL DEFAULT 100"), ("daily_limit", "INTEGER NOT NULL DEFAULT 1000"), ("schedule", "TEXT NOT NULL DEFAULT ''"), ("dry_run", "INTEGER NOT NULL DEFAULT 0"), ("lifecycle", "TEXT NOT NULL DEFAULT 'draft'"), ("paused_at", "TEXT")),
         "source_records": (("discovery_run_id", "INTEGER"), ("normalized_key", "TEXT NOT NULL DEFAULT ''"), ("provenance_json", "TEXT NOT NULL DEFAULT '{}'"), ("response_metadata_json", "TEXT NOT NULL DEFAULT '{}'")),
@@ -107,6 +107,7 @@ def _initialize_database(db_path: str) -> sqlite3.Connection:
         CREATE TABLE sources_rebuilt (
           id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id),
           name TEXT NOT NULL, kind TEXT NOT NULL CHECK(kind IN ('csv','manual','google_places','google_browser_search','bing_local','approved_directory','public_website','permitted_social','ct_logs','dns','rdap')), source_code TEXT NOT NULL DEFAULT '', display_name TEXT NOT NULL DEFAULT '', enabled INTEGER NOT NULL DEFAULT 0, approved INTEGER NOT NULL DEFAULT 0,
+          owner TEXT NOT NULL DEFAULT '', terms_url TEXT NOT NULL DEFAULT '', terms_status TEXT NOT NULL DEFAULT 'unreviewed', rate_limit TEXT NOT NULL DEFAULT '', daily_quota INTEGER, credentials_configured INTEGER NOT NULL DEFAULT 0,
           config_json TEXT NOT NULL DEFAULT '{}', policy_json TEXT NOT NULL DEFAULT '{}', quota_json TEXT NOT NULL DEFAULT '{}', health_status TEXT NOT NULL DEFAULT 'unknown',
           consecutive_failures INTEGER NOT NULL DEFAULT 0, circuit_open INTEGER NOT NULL DEFAULT 0,
           last_success_at TEXT, last_failure_at TEXT, last_error TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
@@ -1449,7 +1450,7 @@ class ApiHandler(BaseHTTPRequestHandler):
             else:seen.add(key);accepted.append(b)
         return self.send_json(200,{"accepted":len(accepted),"duplicates":len(rows)-len(normalized)+len(normalized)-len(accepted)-suppressed,"suppressed":suppressed,"rows":accepted})
     def list_sources(self,db,org):
-        cols='id,organization_id,name,kind,source_code,display_name,enabled,approved,config_json,policy_json,quota_json,health_status,consecutive_failures,circuit_open,last_success_at,last_failure_at,last_error,created_at,updated_at'
+        cols='*'
         items=[]
         adapter_meta={a["source_code"]:a for a in available_adapters()}
         for raw in db.execute(f"SELECT {cols} FROM sources WHERE organization_id=? ORDER BY id",(org,)):
@@ -1468,9 +1469,12 @@ class ApiHandler(BaseHTTPRequestHandler):
                          # back through setup.
                          "configured": bool(meta.get("available", False) and (config.get("csv") or config.get("rows") is not None or raw["approved"] or config.get("approved") is True)),
                          "credential_status": ("Configured" if credential_exists else "Required / not configured") if meta.get("requires_credentials") else "Not required",
-                         "terms_status": "Provided" if policy.get("terms_url") or config.get("terms_url") else "Not reviewed",
-                         "owner": policy.get("owner") or config.get("owner") or "Not assigned",
-                         "rate_limit": policy.get("rate_limit") or config.get("rate_limit") or "Not set"})
+                         "credentials_configured": credential_exists,
+                         "terms_url": raw["terms_url"] or policy.get("terms_url") or config.get("terms_url") or "",
+                         "terms_status": raw["terms_status"] or policy.get("terms_status") or config.get("terms_status") or "unreviewed",
+                         "owner": raw["owner"] or policy.get("owner") or config.get("owner") or "Not assigned",
+                         "rate_limit": raw["rate_limit"] or policy.get("rate_limit") or config.get("rate_limit") or "Not set",
+                         "daily_quota": raw["daily_quota"] if raw["daily_quota"] is not None else quota.get("daily_quota", config.get("daily_quota"))})
             item.pop("config_json", None)
             items.append(item)
         return self.send_json(200,{"organization_id":org,"items":items})
@@ -1507,12 +1511,22 @@ class ApiHandler(BaseHTTPRequestHandler):
         if requested == 'google_browser_search' and bool(payload.get('enabled', False)) and os.environ.get('GOOGLE_BROWSER_SEARCH_ENABLED', '').strip().lower() != 'true': return self.send_json(409,{"error":"source_feature_disabled","detail":"GOOGLE_BROWSER_SEARCH_ENABLED=true is required"})
         if contains_secret(config):return self.send_json(400,{"error":"secret_not_permitted"})
         if any(field in config for field in DISCOVERY_CRITERIA_FIELDS):return self.send_json(400,{"error":"source_configuration_contains_criteria"})
+        policy = payload.get('policy', {}) if isinstance(payload.get('policy', {}), dict) else {}
+        quota = payload.get('quota', {}) if isinstance(payload.get('quota', {}), dict) else {}
+        owner = str(policy.get('owner', config.get('owner', ''))).strip()
+        terms_url = str(policy.get('terms_url', config.get('terms_url', ''))).strip()
+        terms_status = str(policy.get('terms_status', config.get('terms_status', 'unreviewed'))).strip().lower() or 'unreviewed'
+        rate_limit = str(policy.get('rate_limit', config.get('rate_limit', ''))).strip()
+        daily_quota = quota.get('daily_quota', config.get('daily_quota'))
+        try: daily_quota = int(daily_quota) if daily_quota not in (None, '') else None
+        except (TypeError, ValueError): return self.send_json(400, {'error':'invalid_source_quota'})
+        if daily_quota is not None and not 1 <= daily_quota <= 100000: return self.send_json(400, {'error':'invalid_source_quota'})
         try:
             validation=adapter_for(source_code).validate_config(config)
             # Registration may precede the actual local payload or optional provider
             # approval. Discovery/ingest still validates the effective configuration.
             if config and not validation.valid:return self.send_json(400,{"error":"invalid_source_config","details":validation.errors})
-            cur=db.execute("INSERT INTO sources(organization_id,name,kind,source_code,display_name,enabled,approved,config_json,policy_json,quota_json) VALUES(?,?,?,?,?,?,?,?,?,?)",(user['organization_id'],name,kind,source_code,str(payload.get('display_name') or adapter_for(source_code).display_name),int(bool(payload.get('enabled',False))),int(bool(payload.get('approved',config.get('approved',False)))),json.dumps(config,sort_keys=True),json.dumps(payload.get('policy',{}),sort_keys=True),json.dumps(payload.get('quota',{}),sort_keys=True)))
+            cur=db.execute("INSERT INTO sources(organization_id,name,kind,source_code,display_name,enabled,approved,owner,terms_url,terms_status,rate_limit,daily_quota,credentials_configured,config_json,policy_json,quota_json) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",(user['organization_id'],name,kind,source_code,str(payload.get('display_name') or adapter_for(source_code).display_name),int(bool(payload.get('enabled',False))),int(bool(payload.get('approved',config.get('approved',False)))),owner,terms_url,terms_status,rate_limit,daily_quota,0,json.dumps(config,sort_keys=True),json.dumps(policy,sort_keys=True),json.dumps(quota,sort_keys=True)))
         except sqlite3.IntegrityError:
             existing=db.execute("SELECT id,organization_id,name,kind,source_code,display_name,enabled,approved,policy_json,quota_json,health_status,consecutive_failures,circuit_open,last_success_at,last_failure_at,last_error,created_at,updated_at FROM sources WHERE organization_id=? AND name=?",(user['organization_id'],name)).fetchone()
             if existing:
diff --git a/apps/api/schema.sql b/apps/api/schema.sql
index bfea3af..5ea8db9 100644
--- a/apps/api/schema.sql
+++ b/apps/api/schema.sql
@@ -137,6 +137,7 @@ CREATE INDEX IF NOT EXISTS idx_job_events_job_sequence ON job_events(job_id,sequ
 CREATE TABLE IF NOT EXISTS sources (
  id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id),
  name TEXT NOT NULL, kind TEXT NOT NULL CHECK(kind IN ('csv','manual','google_places','google_browser_search','bing_local','approved_directory','public_website','permitted_social','ct_logs','dns','rdap')), source_code TEXT NOT NULL DEFAULT '', display_name TEXT NOT NULL DEFAULT '', enabled INTEGER NOT NULL DEFAULT 0, approved INTEGER NOT NULL DEFAULT 0,
+ owner TEXT NOT NULL DEFAULT '', terms_url TEXT NOT NULL DEFAULT '', terms_status TEXT NOT NULL DEFAULT 'unreviewed', rate_limit TEXT NOT NULL DEFAULT '', daily_quota INTEGER, credentials_configured INTEGER NOT NULL DEFAULT 0,
  config_json TEXT NOT NULL DEFAULT '{}', policy_json TEXT NOT NULL DEFAULT '{}', quota_json TEXT NOT NULL DEFAULT '{}', health_status TEXT NOT NULL DEFAULT 'unknown',
  consecutive_failures INTEGER NOT NULL DEFAULT 0, circuit_open INTEGER NOT NULL DEFAULT 0,
  last_success_at TEXT, last_failure_at TEXT, last_error TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
diff --git a/apps/api/tests/test_sources_phase5.py b/apps/api/tests/test_sources_phase5.py
index 9b91372..ae2a485 100644
--- a/apps/api/tests/test_sources_phase5.py
+++ b/apps/api/tests/test_sources_phase5.py
@@ -113,6 +113,24 @@ class SourceApiTests(unittest.TestCase):
         second={**payload,'name':'OpenStreetMap / Overpass · plumbers · Durban'}
         status, other=self.req('POST','/api/v1/sources',second); self.assertEqual(status,201); self.assertTrue(other['created']); self.assertNotEqual(other['id'],created['id'])
 
+    def test_source_registry_exposes_first_class_governance_fields(self):
+        payload = {
+            'name': 'Governed manual', 'kind': 'manual', 'config': {'rows': []},
+            'policy': {'owner': 'Research Ops', 'terms_url': 'https://example.test/terms', 'terms_status': 'approved', 'rate_limit': '60/hour'},
+            'quota': {'daily_quota': 200},
+        }
+        status, created = self.req('POST', '/api/v1/sources', payload)
+        self.assertEqual(status, 201)
+        status, items = self.req('GET', '/api/v1/sources')
+        self.assertEqual(status, 200)
+        source = next(item for item in items['items'] if item['id'] == created['id'])
+        self.assertEqual(source['owner'], 'Research Ops')
+        self.assertEqual(source['terms_url'], 'https://example.test/terms')
+        self.assertEqual(source['terms_status'], 'approved')
+        self.assertEqual(source['rate_limit'], '60/hour')
+        self.assertEqual(source['daily_quota'], 200)
+        self.assertFalse(source['credentials_configured'])
+
     def test_source_configuration_rejects_discovery_criteria(self):
         forbidden = {'query': 'plumbers', 'category': 'trades', 'city': 'Cape Town', 'location': 'Western Cape'}
         for field, value in forbidden.items():