diff --git a/pkg/duckdb_parser/builder.go b/pkg/duckdb_parser/builder.go
index ab2c7a1df..90e930f22 100644
--- a/pkg/duckdb_parser/builder.go
+++ b/pkg/duckdb_parser/builder.go
@@ -41,7 +41,8 @@ func NewBuilder() *QueryBuilder {
 }
 
 type ingestParams struct {
-	FilePath string
+	FilePath     string
+	CollectionID int64
 }
 
 // CreateSchemaQuery returns queries to create all RVTools tables with proper schema.
@@ -55,8 +56,20 @@ func (b *QueryBuilder) IngestRvtoolsQuery(filePath string) (string, error) {
 }
 
 // IngestSqliteQuery returns a query that creates RVTools-shaped tables from a forklift SQLite database.
+// Use IngestSqliteQueryWithCollection to hash VM IDs and stamp collection_id.
 func (b *QueryBuilder) IngestSqliteQuery(filePath string) (string, error) {
-	return b.buildQuery("ingest_sqlite", mustGetTemplate("ingest_sqlite"), ingestParams{FilePath: filePath})
+	return b.IngestSqliteQueryWithCollection(filePath, 0)
+}
+
+// IngestSqliteQueryWithCollection returns the same query as IngestSqliteQuery but, when
+// collectionID > 0, hashes each VM's "VM ID" to md5('{collectionID}_{vSphere_MOID}'),
+// stores the original MOID in the vmmoid column, and writes collection_id.
+func (b *QueryBuilder) IngestSqliteQueryWithCollection(filePath string, collectionID int64) (string, error) {
+	if collectionID < 0 {
+		return "", fmt.Errorf("collectionID must be non-negative, got %d", collectionID)
+	}
+	return b.buildQuery("ingest_sqlite", mustGetTemplate("ingest_sqlite"),
+		ingestParams{FilePath: filePath, CollectionID: collectionID})
 }
 
 // queryParams holds all template parameters for queries.
diff --git a/pkg/duckdb_parser/ingest.go b/pkg/duckdb_parser/ingest.go
index cec3b4b15..4034f9cd1 100644
--- a/pkg/duckdb_parser/ingest.go
+++ b/pkg/duckdb_parser/ingest.go
@@ -27,6 +27,12 @@ func isCriticalStatement(stmt string) bool {
 			return true
 		}
 	}
+	// INSERT INTO vinfo is critical only when it carries collection_id/vmmoid columns
+	// (SQLite collection path). The RVTools INSERT INTO vinfo does not include these columns
+	// and may legitimately fail when the source Excel is malformed.
+	if strings.Contains(upperStmt, "INSERT INTO VINFO") && strings.Contains(upperStmt, "VMMOID") {
+		return true
+	}
 	return false
 }
 
@@ -81,12 +87,23 @@ func (p *Parser) IngestRvTools(ctx context.Context, excelFile string) (Validatio
 	return result, nil
 }
 
-// IngestSqlite ingests data from a forklift SQLite database, runs VM validation if a validator
-// is configured, and validates the schema for required tables/columns.
-// Returns a ValidationResult with errors (fatal) and warnings (non-fatal).
-// If ValidationResult.HasErrors() is true, the inventory cannot be built.
+// IngestSqlite ingests data from a forklift SQLite database with no collection context.
+// Existing behaviour is preserved exactly. Use IngestSqliteWithCollection when the
+// agent has an active collection and wants hashed VM IDs.
 func (p *Parser) IngestSqlite(ctx context.Context, sqliteFile string) (ValidationResult, error) {
-	query, err := p.builder.IngestSqliteQuery(sqliteFile)
+	return p.IngestSqliteWithCollection(ctx, sqliteFile, 0)
+}
+
+// IngestSqliteWithCollection ingests from a forklift SQLite database.
+// When collectionID > 0 every "VM ID" in vinfo and relational tables is replaced
+// with md5('{collectionID}_{original_moid}'); the original MOID is written to
+// the vmmoid column and collection_id is set. When collectionID = 0 the
+// behaviour is identical to IngestSqlite (no hashing, no extra columns written).
+func (p *Parser) IngestSqliteWithCollection(ctx context.Context, sqliteFile string, collectionID int64) (ValidationResult, error) {
+	if collectionID < 0 {
+		return ValidationResult{}, fmt.Errorf("collectionID must be non-negative, got %d", collectionID)
+	}
+	query, err := p.builder.IngestSqliteQueryWithCollection(sqliteFile, collectionID)
 	if err != nil {
 		return ValidationResult{}, fmt.Errorf("building sqlite ingestion query: %w", err)
 	}
@@ -94,10 +111,8 @@ func (p *Parser) IngestSqlite(ctx context.Context, sqliteFile string) (Validatio
 		return ValidationResult{}, fmt.Errorf("ingesting sqlite data: %w", err)
 	}
 
-	// Validate schema against vinfo (SQLite inserts directly into vinfo, no vinfo_raw)
 	result := p.ValidateSchema(ctx, "vinfo")
 
-	// Only run post-ingestion steps if schema is valid (we have VMs to process)
 	if result.IsValid() {
 		if err := p.populateComplexity(ctx); err != nil {
 			return result, fmt.Errorf("populating complexity: %w", err)
diff --git a/pkg/duckdb_parser/ingest_sqlite_test.go b/pkg/duckdb_parser/ingest_sqlite_test.go
index b9508113c..71193560f 100644
--- a/pkg/duckdb_parser/ingest_sqlite_test.go
+++ b/pkg/duckdb_parser/ingest_sqlite_test.go
@@ -2,6 +2,7 @@ package duckdb_parser
 
 import (
 	"context"
+	"crypto/md5"
 	"database/sql"
 	"fmt"
 	"path/filepath"
@@ -389,3 +390,134 @@ func TestIngestSqlite_PopulateVCluster_SkipsWhenAlreadyPopulated(t *testing.T) {
 	require.NoError(t, db.QueryRowContext(ctx, `SELECT "Object ID" FROM vcluster WHERE "Name" = 'existing-cluster'`).Scan(&objectID))
 	assert.Equal(t, "sentinel-id", objectID, "pre-existing row must not be overwritten")
 }
+
+// addCollectionColumns adds vmmoid and collection_id columns to vinfo so that
+// IngestSqliteWithCollection can write to them. In production these columns are
+// added by agent migration 026; in tests we add them manually.
+func addCollectionColumns(t *testing.T, db *sql.DB) {
+	t.Helper()
+	ctx := context.Background()
+	_, err := db.ExecContext(ctx, `ALTER TABLE vinfo ADD COLUMN IF NOT EXISTS vmmoid VARCHAR`)
+	require.NoError(t, err, "adding vmmoid column to vinfo")
+	_, err = db.ExecContext(ctx, `ALTER TABLE vinfo ADD COLUMN IF NOT EXISTS collection_id BIGINT`)
+	require.NoError(t, err, "adding collection_id column to vinfo")
+}
+
+func TestIngestSqliteWithCollection_HashesVMIDsAndSetsVmmoid(t *testing.T) {
+	const collectionID = int64(7)
+	ctx := context.Background()
+
+	parser, db, cleanup := setupTestParser(t, &testValidator{})
+	defer cleanup()
+
+	addCollectionColumns(t, db)
+
+	clusters := []sqliteCluster{
+		{id: "domain-c1", name: "cluster1", datacenter: "dc1"},
+	}
+	vms := []sqliteVM{
+		{id: "vm-001", name: "vm-1", clusterName: "cluster1"},
+		{id: "vm-002", name: "vm-2", clusterName: "cluster1"},
+	}
+	sqlitePath := createTestSQLite(t, "vcenter-uuid-001", clusters, vms)
+
+	result, err := parser.IngestSqliteWithCollection(ctx, sqlitePath, collectionID)
+	require.NoError(t, err)
+	require.True(t, result.IsValid())
+
+	rows, err := db.QueryContext(ctx, `SELECT "VM ID", vmmoid, collection_id FROM vinfo`)
+	require.NoError(t, err)
+	defer func() { _ = rows.Close() }()
+
+	rowCount := 0
+	for rows.Next() {
+		rowCount++
+		var vmID, vmmoid string
+		var colID int64
+		require.NoError(t, rows.Scan(&vmID, &vmmoid, &colID))
+
+		// "VM ID" must be a 32-char lowercase hex md5 string.
+		assert.Len(t, vmID, 32, "VM ID should be a 32-char md5 hex string")
+		assert.Regexp(t, `^[0-9a-f]{32}$`, vmID, "VM ID should be a lowercase hex string")
+
+		// vmmoid is the original MOID (non-empty, not itself a 32-char hash).
+		assert.NotEmpty(t, vmmoid, "vmmoid should be the original MOID")
+		assert.NotEqual(t, 32, len(vmmoid), "vmmoid should not look like an md5 hash (original MOIDs are shorter)")
+
+		// collection_id must be set correctly.
+		assert.Equal(t, collectionID, colID, "collection_id should match the supplied collectionID")
+
+		// Hash must be deterministic: md5("{collectionID}_{vmmoid}") == vmID
+		expectedHash := fmt.Sprintf("%x", md5.Sum(fmt.Appendf(nil, "%d_%s", collectionID, vmmoid)))
+		assert.Equal(t, expectedHash, vmID, "VM ID should equal md5('%d_%s', collectionID, vmmoid)")
+	}
+	require.NoError(t, rows.Err())
+	assert.Equal(t, len(vms), rowCount, "vinfo should have one row per non-template VM")
+}
+
+func TestIngestSqliteWithCollection_RelationalTablesMatchVinfoIDs(t *testing.T) {
+	const collectionID = int64(7)
+	ctx := context.Background()
+
+	parser, db, cleanup := setupTestParser(t, &testValidator{})
+	defer cleanup()
+
+	addCollectionColumns(t, db)
+
+	clusters := []sqliteCluster{
+		{id: "domain-c1", name: "cluster1", datacenter: "dc1"},
+	}
+	vms := []sqliteVM{
+		{id: "vm-001", name: "vm-1", clusterName: "cluster1"},
+		{id: "vm-002", name: "vm-2", clusterName: "cluster1"},
+	}
+	sqlitePath := createTestSQLite(t, "vcenter-uuid-001", clusters, vms)
+
+	_, err := parser.IngestSqliteWithCollection(ctx, sqlitePath, collectionID)
+	require.NoError(t, err)
+
+	// Every vcpu "VM ID" must exist in vinfo "VM ID" — no orphaned rows.
+	var orphans int
+	err = db.QueryRowContext(ctx, `
+		SELECT COUNT(*) FROM vcpu
+		WHERE "VM ID" NOT IN (SELECT "VM ID" FROM vinfo)
+	`).Scan(&orphans)
+	require.NoError(t, err)
+	assert.Equal(t, 0, orphans, "all vcpu rows should reference a valid vinfo VM ID")
+}
+
+func TestIngestSqliteWithCollection_ZeroCollectionIDMatchesIngestSqlite(t *testing.T) {
+	ctx := context.Background()
+
+	clusters := []sqliteCluster{
+		{id: "domain-c1", name: "cluster1", datacenter: "dc1"},
+	}
+	vms := []sqliteVM{
+		{id: "vm-001", name: "vm-1", clusterName: "cluster1"},
+		{id: "vm-002", name: "vm-2", clusterName: "cluster1"},
+	}
+
+	// Parser A uses IngestSqlite (the zero-collection-ID path).
+	parserA, dbA, cleanupA := setupTestParser(t, &testValidator{})
+	defer cleanupA()
+
+	// Parser B uses IngestSqliteWithCollection with collectionID=0.
+	parserB, dbB, cleanupB := setupTestParser(t, &testValidator{})
+	defer cleanupB()
+
+	sqlitePathA := createTestSQLite(t, "vcenter-uuid-001", clusters, vms)
+	sqlitePathB := createTestSQLite(t, "vcenter-uuid-001", clusters, vms)
+
+	resultA, err := parserA.IngestSqlite(ctx, sqlitePathA)
+	require.NoError(t, err)
+
+	resultB, err := parserB.IngestSqliteWithCollection(ctx, sqlitePathB, 0)
+	require.NoError(t, err)
+
+	assert.Equal(t, resultA.IsValid(), resultB.IsValid(), "validity should match between IngestSqlite and IngestSqliteWithCollection(0)")
+
+	var countA, countB int
+	require.NoError(t, dbA.QueryRowContext(ctx, `SELECT COUNT(*) FROM vinfo`).Scan(&countA))
+	require.NoError(t, dbB.QueryRowContext(ctx, `SELECT COUNT(*) FROM vinfo`).Scan(&countB))
+	assert.Equal(t, countA, countB, "VM count should match between IngestSqlite and IngestSqliteWithCollection(0)")
+}
diff --git a/pkg/duckdb_parser/templates/ingest_sqlite.go.tmpl b/pkg/duckdb_parser/templates/ingest_sqlite.go.tmpl
index d97ade01a..80fdc1997 100644
--- a/pkg/duckdb_parser/templates/ingest_sqlite.go.tmpl
+++ b/pkg/duckdb_parser/templates/ingest_sqlite.go.tmpl
@@ -18,7 +18,7 @@ Key transformations:
 ATTACH '{{.FilePath}}' AS src (TYPE sqlite);
 
 INSERT INTO vinfo (
-    "VM ID", "VM", "Folder ID", "Folder", "Host", "SMBIOS UUID", "VM UUID",
+    "VM ID", {{if .CollectionID}}"vmmoid", "collection_id", {{end}}"VM", "Folder ID", "Folder", "Host", "SMBIOS UUID", "VM UUID",
     "Firmware", "Powerstate", "Connection state", "FT State",
     "CPUs", "Memory",
     "OS according to the configuration file", "OS according to the VMware Tools",
@@ -29,7 +29,7 @@ INSERT INTO vinfo (
     "migration_excluded", "labels", "guest_apps"
 )
 SELECT
-    v.ID,
+    {{if .CollectionID}}md5(CAST({{.CollectionID}} AS VARCHAR) || '_' || v.ID), v.ID, {{.CollectionID}},{{else}}v.ID,{{end}}
     v.Name,
     v.Folder,
     v.Folder,
@@ -70,21 +70,21 @@ WHERE v.IsTemplate = 0;
 
 INSERT INTO vcpu ("VM ID", "Hot Add", "Hot Remove", "Sockets", "Cores p/s")
 SELECT
-    v.ID,
+    {{if .CollectionID}}md5(CAST({{.CollectionID}} AS VARCHAR) || '_' || v.ID){{else}}v.ID{{end}},
     v.CpuHotAddEnabled = 1,
     v.CpuHotRemoveEnabled = 1,
     v.CpuCount / NULLIF(v.CoresPerSocket, 0),
     v.CoresPerSocket
 FROM src.VM v
-WHERE v.ID IN (SELECT "VM ID" FROM vinfo);
+WHERE v.ID IN (SELECT {{if .CollectionID}}vmmoid{{else}}"VM ID"{{end}} FROM vinfo{{if .CollectionID}} WHERE vmmoid IS NOT NULL{{end}});
 
 INSERT INTO vmemory ("VM ID", "Hot Add", "Ballooned")
 SELECT
-    v.ID,
+    {{if .CollectionID}}md5(CAST({{.CollectionID}} AS VARCHAR) || '_' || v.ID){{else}}v.ID{{end}},
     v.MemoryHotAddEnabled = 1,
     v.BalloonedMemory
 FROM src.VM v
-WHERE v.ID IN (SELECT "VM ID" FROM vinfo);
+WHERE v.ID IN (SELECT {{if .CollectionID}}vmmoid{{else}}"VM ID"{{end}} FROM vinfo{{if .CollectionID}} WHERE vmmoid IS NOT NULL{{end}});
 
 INSERT INTO vdisk (
     "VM ID", "Disk Key", "Unit #", "Path", "Disk Path", "Capacity MiB",
@@ -92,7 +92,7 @@ INSERT INTO vdisk (
     "Thin", "Controller", "Label", "SCSI Unit #"
 )
 SELECT
-    v.ID,
+    {{if .CollectionID}}md5(CAST({{.CollectionID}} AS VARCHAR) || '_' || v.ID){{else}}v.ID{{end}},
     disk->>'key',
     disk->>'unitNumber',
     disk->>'file',
@@ -109,11 +109,11 @@ SELECT
     disk->>'unitNumber'
 FROM src.VM v,
 LATERAL unnest(from_json(v.Disks, '[{"key":"INTEGER","unitNumber":"INTEGER","file":"VARCHAR","capacity":"BIGINT","shared":"BOOLEAN","rdm":"BOOLEAN","bus":"VARCHAR","mode":"VARCHAR","serial":"VARCHAR"}]')) AS t(disk)
-WHERE v.Disks != '[]' AND v.ID IN (SELECT "VM ID" FROM vinfo);
+WHERE v.Disks != '[]' AND v.ID IN (SELECT {{if .CollectionID}}vmmoid{{else}}"VM ID"{{end}} FROM vinfo{{if .CollectionID}} WHERE vmmoid IS NOT NULL{{end}});
 
 WITH nic_flat AS (
     SELECT
-        v.ID AS vm_id,
+        {{if .CollectionID}}md5(CAST({{.CollectionID}} AS VARCHAR) || '_' || v.ID){{else}}v.ID{{end}} AS vm_id,
         v.IpAddress,
         v.GuestNetworks,
         nic.mac AS mac_addr,
@@ -122,7 +122,7 @@ WITH nic_flat AS (
     FROM src.VM v
     LEFT JOIN src.Host h ON v.Host = h.ID,
     LATERAL unnest(from_json(v.NICs, '[{"network":{"kind":"VARCHAR","id":"VARCHAR"},"mac":"VARCHAR","order":"INTEGER","deviceKey":"INTEGER"}]')) AS t(nic)
-    WHERE v.NICs != '[]' AND v.ID IN (SELECT "VM ID" FROM vinfo)
+    WHERE v.NICs != '[]' AND v.ID IN (SELECT {{if .CollectionID}}vmmoid{{else}}"VM ID"{{end}} FROM vinfo{{if .CollectionID}} WHERE vmmoid IS NOT NULL{{end}})
 ),
 gn_flat AS (
     SELECT